diff --git a/AGENTS.md b/AGENTS.md
index 00a48d90..3d23774e 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -75,12 +75,15 @@ Do not:
## Local Development
+Event codec: `CODEC_PROVIDER=json` is required wherever workers are exercised (the worker command/result envelopes carry untyped bodies Avro cannot serialize); the Makefile and docker-compose set it everywhere. `tracking-events` keeps its own Avro path regardless.
+
Infra runs in docker; the Go services and frontends run natively on the host for fast iteration — no docker image rebuilds when you change app code. Targets live in the `Makefile`.
- `make infra` — start the backing services in docker (postgres, redis, kafka, schema-registry, mailpit, localstack + init, cloud-tasks, stripe-mock). Run once; leave running.
- `make backend` — run the API natively on `:8080` (applies the embedded migrations on boot against the docker postgres).
-- `make consumer` / `make worker` — run those Go services natively, each in its own terminal. The worker reads encrypted DEKs through the backend's `/internal/dek` endpoint (the prod `http` provider, no worker DB), so `make backend` must be running and their `INTERNAL_API_TOKEN` must match (the targets are pre-wired to match).
-- `make run` — backend + consumer + worker together in one terminal (Ctrl-C stops all).
+- `make consumer` / `make worker` / `make worker-premium` — run those Go services natively, each in its own terminal. Two native workers exist because tier placement is strict: free-trial orgs place onto the free-tier worker (`make worker`), paid orgs onto the premium one (`make worker-premium`). The workers read encrypted DEKs through the backend's `/internal/dek` endpoint (the prod `http` provider, no worker DB), so `make backend` must be running and their `INTERNAL_API_TOKEN` must match (the targets are pre-wired to match).
+- `make run` — backend + consumer + both workers together in one terminal (Ctrl-C stops all).
+- `make sandbox` — fully working demo environment: seeds the "Sunrise Labs" showcase org (live mailboxes: SMTP -> mailpit, IMAP -> dovecot, credentials sealed with `CREDENTIALS_ENCRYPTION_KEY`) and runs the simulator that plays the internet (delivers mail into dovecot inboxes, opens pixels, clicks tracked links, replies as contacts). Needs `make run` + `make tracking` alongside. Docs: `docs/content/docs/development/sandbox.mdx`.
- `make tracking` / `make realtime` — the Rust tracking pixel service (:3000) and Elixir/Phoenix websocket fanout (:4000). Deliberately kept out of `make run`; start them only when needed, and only if you have the cargo / elixir toolchains on the host.
- `make web` / `make admin` / `make site` — frontend dev servers (5173 / 5174 / 4321), pointed at the native backend.
- `make seed` — load fixtures (after the backend has applied migrations).
diff --git a/Makefile b/Makefile
index b2f8e549..6847c8a5 100644
--- a/Makefile
+++ b/Makefile
@@ -23,9 +23,9 @@ PROTO_DIR := internal/tasks/proto
PROTO_GEN_FILES := $(PROTO_DIR)/tasks.pb.go
.PHONY: setup-tools fmt lint proto check-proto \
- up sim seed seed-plan reset logs status stop down tools test-seed \
+ up sim seed seed-plan sandbox sandbox-seed reset logs status stop down tools test-seed \
restart restart-go restart-all infra infra-down app app-down app-logs \
- backend consumer worker run tracking realtime web \
+ backend consumer worker worker-premium run tracking realtime web \
admin site docs grant-admin revoke-admin
setup-tools:
@@ -71,6 +71,19 @@ up:
sim:
$(COMPOSE) --profile sim up
+# Fully working demo environment: seeds the "Sunrise Labs" showcase org
+# (live mailboxes on mailpit/dovecot, active campaigns, warmup pool, pro
+# plan) and then runs the simulator that plays the internet: delivering
+# captured mail into recipient inboxes, opening pixels, clicking tracked
+# links, and replying as the seeded contacts. Requires the rest of the
+# stack: `make infra` + `make run` + `make tracking` (and `make realtime`
+# + `make web` for the live dashboard). Docs: /development/sandbox/.
+sandbox:
+ $(GO_DEV_ENV) go run ./cmd/sandbox
+
+sandbox-seed:
+ $(GO_DEV_ENV) go run ./cmd/sandbox -seed-only
+
# Load rich fixture data. Runs natively like the other dev services — the
# seeder only needs Postgres, so it does not depend on a (re)built docker
# backend image, just `make infra` plus migrations applied (`make migrate`,
@@ -208,8 +221,8 @@ DEV_COMPOSE := $(COMPOSE) -f docker-compose.yml -f docker-compose.dev.yml
# bucket. localstack runs with PERSISTENCE=0, so those are wiped on every
# restart and must be recreated before any service (incl. the natively-run
# backend) touches KMS/S3.
-INFRA_SVCS := postgres redis zookeeper kafka kafka-init schema-registry \
- mailpit localstack localstack-init stripe-mock cloud-tasks-emulator
+INFRA_SVCS := postgres redis zookeeper kafka kafka-init schema-registry schema-registry-init \
+ mailpit dovecot localstack localstack-init stripe-mock cloud-tasks-emulator
# Language services. The things you iterate on; recreated per worktree.
APP_SVCS := backend consumer worker-shared-1 tracking realtime web
@@ -283,7 +296,12 @@ app-logs:
# make run INFRA_HOST=192.168.1.50 SELF_HOST=192.168.1.42
#
INFRA_HOST ?= localhost
-SELF_HOST ?= localhost
+# SELF_HOST is how the DOCKERIZED cloud-tasks emulator reaches the natively
+# running backend; from inside a container "localhost" is the container
+# itself, so the Docker Desktop host alias is the working default (the
+# emulator's compose service adds a host-gateway mapping so it also resolves
+# on Linux). Override when the backend runs on another machine.
+SELF_HOST ?= host.docker.internal
# ─── expose the dev servers off-box (Tailscale / LAN) ───────────────────
#
@@ -325,8 +343,18 @@ CORS_ORIGINS := $(if $(PUBLIC_HOST),http://$(PUBLIC_HOST):5173$(comma)http://$(P
# Shared by the control-plane services (backend, consumer). Flattened to
# one line by make so it can prefix a command as inline env.
+# Fixed dev key sealing SMTP/IMAP credentials at rest (64 hex chars). The
+# backend/consumer decrypt with it and cmd/sandbox seeds with it, so all
+# three must share the value. Never reuse in production.
+CREDENTIALS_KEY_DEV := 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
+
+# CODEC_PROVIDER=json: the worker command/result envelopes carry `any`
+# bodies that Avro cannot serialize, so worker messaging only works on the
+# JSON codec. tracking-events stays Avro (dedicated Avrov2 path).
GO_DEV_ENV := \
APP_ENV=dev \
+ CODEC_PROVIDER=json \
+ CREDENTIALS_ENCRYPTION_KEY=$(CREDENTIALS_KEY_DEV) \
AWS_CONFIG_ENABLED=false \
AWS_ENDPOINT_URL=http://$(INFRA_HOST):4566 \
AWS_REGION=us-east-1 \
@@ -347,6 +375,8 @@ GO_DEV_ENV := \
# — it has no relational access by design.
WORKER_DEV_ENV := \
APP_ENV=dev \
+ CODEC_PROVIDER=json \
+ MAIL_TLS_INSECURE=true \
AWS_CONFIG_ENABLED=false \
AWS_ENDPOINT_URL=http://$(INFRA_HOST):4566 \
AWS_REGION=us-east-1 \
@@ -414,14 +444,27 @@ worker:
ENCRYPTED_KEYS_WORKER_TOKEN=local-dev-internal-token \
go run ./cmd/worker
+# The premium shared worker (seeded as premium-1, free_tier=false). Paid
+# orgs place strictly onto premium workers, so without this one running
+# natively their mailboxes never send or sync (the sandbox org needs it).
+worker-premium:
+ $(WORKER_DEV_ENV) \
+ WORKER_ID=10c8f5e4-1c39-5b2a-9c8b-3d2f0a8b1a02 \
+ WORKER_TIER=shared \
+ ENCRYPTED_KEYS_PROVIDER=http \
+ ENCRYPTED_KEYS_BACKEND_URL=http://localhost:8080 \
+ ENCRYPTED_KEYS_WORKER_TOKEN=local-dev-internal-token \
+ go run ./cmd/worker
+
# backend + consumer + worker together in one terminal. Ctrl-C stops all
# (kill 0 takes down go run and its child binaries). Run `make infra` first.
run:
- @echo "backend + consumer + worker (native). Ctrl-C stops all. Run 'make infra' first if infra is down."
+ @echo "backend + consumer + both shared workers (native). Ctrl-C stops all. Run 'make infra' first if infra is down."
@trap 'kill 0' INT TERM; \
$(MAKE) --no-print-directory backend & \
$(MAKE) --no-print-directory consumer & \
$(MAKE) --no-print-directory worker & \
+ $(MAKE) --no-print-directory worker-premium & \
wait
# ─── other native services (Rust tracking, Elixir realtime) ──────────────
diff --git a/cmd/backend/main.go b/cmd/backend/main.go
index 6e07c68a..dfe18740 100644
--- a/cmd/backend/main.go
+++ b/cmd/backend/main.go
@@ -90,6 +90,7 @@ import (
"github.com/warmbly/warmbly/internal/observability"
"github.com/warmbly/warmbly/internal/pkg/captcha"
"github.com/warmbly/warmbly/internal/pkg/emailverify"
+ "github.com/warmbly/warmbly/internal/pkg/encrypt"
"github.com/warmbly/warmbly/internal/pkg/generation"
"github.com/warmbly/warmbly/internal/pkg/geo"
"github.com/warmbly/warmbly/internal/pkg/idtoken"
@@ -431,9 +432,17 @@ func main() {
log.Fatal(err)
}
- // Codec wraps the same Avrov2 client so EventBus payloads decode the
- // same way regardless of transport.
- codecImpl := codec.NewAvroFromClient(avrov2Client)
+ // Codec for EventBus payloads. Must match the worker fleet's
+ // CODEC_PROVIDER: the worker command envelopes (WorkerEvent/JobEvent
+ // carry `any` bodies) only serialize as JSON, so json is required
+ // wherever workers are exercised; avro remains the default for the
+ // historical Schema Registry topics.
+ var codecImpl codec.Codec
+ if os.Getenv("CODEC_PROVIDER") == "json" {
+ codecImpl = codec.NewJSON()
+ } else {
+ codecImpl = codec.NewAvroFromClient(avrov2Client)
+ }
// Legacy Kafka producer still used by email + tasks services that
// haven't been migrated to EventBus yet. Removing this is follow-up
@@ -480,7 +489,12 @@ func main() {
authRepostory := repository.NewAuthRepostory(primaryDB)
tokenRepostory := repository.NewTokenRepostory(primaryDB)
webauthnRepository := repository.NewWebAuthnRepository(primaryDB)
- emailRepostory := repository.NewEmailRepostory(primaryDB)
+ credEncrypter, cerr := encrypt.FromEnv()
+ if cerr != nil {
+ sentry.CaptureException(cerr)
+ log.Fatal("Invalid CREDENTIALS_ENCRYPTION_KEY: ", cerr)
+ }
+ emailRepostory := repository.NewEmailRepostory(primaryDB, credEncrypter)
campaignRepostory := repository.NewCampaignRepostory(primaryDB)
sequenceRepostory := repository.NewSequenceRepostory(primaryDB)
contactRepostory := repository.NewContactRepostory(primaryDB)
diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go
index 42967fee..d1322e90 100644
--- a/cmd/consumer/main.go
+++ b/cmd/consumer/main.go
@@ -32,6 +32,7 @@ import (
"github.com/warmbly/warmbly/internal/infrastructure/storage"
"github.com/warmbly/warmbly/internal/notify"
"github.com/warmbly/warmbly/internal/observability"
+ "github.com/warmbly/warmbly/internal/pkg/encrypt"
"github.com/warmbly/warmbly/internal/repository"
)
@@ -182,7 +183,12 @@ func main() {
}
// Repositories
- emailRepo := repository.NewEmailRepostory(primaryDB)
+ credEncrypter, err := encrypt.FromEnv()
+ if err != nil {
+ sentry.CaptureException(err)
+ log.Fatal("Invalid CREDENTIALS_ENCRYPTION_KEY: ", err)
+ }
+ emailRepo := repository.NewEmailRepostory(primaryDB, credEncrypter)
uniboxRepo := repository.NewUniboxRepository(primaryDB)
mailboxRepo := repository.NewMailboxRepository(primaryDB)
emailHistoryIDRepo := repository.NewEmailHistoryIDRepository(primaryDB)
@@ -289,12 +295,21 @@ func main() {
Bootstrap: kafkaBootstrapServers,
SASL: kafkaSaslConfig,
})
- consumerCodec := codec.NewAvroFromClient(avrov2Client)
+ // Same CODEC_PROVIDER contract as backend/worker: the worker event
+ // envelopes only serialize as JSON. tracking-events stays on its own
+ // Avro deserializer (the Rust producer always writes Avro).
+ var consumerCodec codec.Codec
+ if os.Getenv("CODEC_PROVIDER") == "json" {
+ consumerCodec = codec.NewJSON()
+ } else {
+ consumerCodec = codec.NewAvroFromClient(avrov2Client)
+ }
eventsPublisher := events.NewPublisher(consumerBus, s3Client, consumerCodec, cipherService)
// JobsService
jobsService := &jobs.JobsService{
Consumer: kafkaConsumer,
+ Codec: consumerCodec,
UniboxRepository: uniboxRepo,
MailboxRepository: mailboxRepo,
EmailRepository: emailRepo,
diff --git a/cmd/sandbox/main.go b/cmd/sandbox/main.go
new file mode 100644
index 00000000..78bebc9a
--- /dev/null
+++ b/cmd/sandbox/main.go
@@ -0,0 +1,66 @@
+// Sandbox: a fully working local demo environment.
+//
+// Seeds a paid showcase org ("Sunrise Labs") whose mailboxes really send
+// (SMTP -> mailpit) and really sync (IMAP <- dovecot), then runs a simulator
+// that plays the internet: routing captured mail into recipient inboxes,
+// opening tracking pixels, clicking tracked links, and replying as the seeded
+// contacts. The platform side (scheduler, worker, consumer, tracking,
+// realtime) is all production code; run it with `make infra` + `make run` +
+// `make tracking` (+ `make realtime` + `make web` for the live dashboard).
+//
+// make sandbox # seed + simulate (foreground)
+// go run ./cmd/sandbox -seed-only
+// go run ./cmd/sandbox -simulate-only
+//
+// Documented at docs.warmbly.com/development/sandbox/.
+package main
+
+import (
+ "context"
+ "flag"
+ "log"
+ "os/signal"
+ "syscall"
+ "time"
+
+ "github.com/jackc/pgx/v5/pgxpool"
+ "github.com/warmbly/warmbly/internal/infrastructure/db"
+ "github.com/warmbly/warmbly/internal/sandbox"
+)
+
+func main() {
+ seedOnly := flag.Bool("seed-only", false, "seed the sandbox org and exit")
+ simulateOnly := flag.Bool("simulate-only", false, "skip seeding, run only the simulator")
+ flag.Parse()
+
+ ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+ defer stop()
+
+ cfg := sandbox.FromEnv()
+
+ pool, err := pgxpool.New(ctx, cfg.DatabaseURL)
+ if err != nil {
+ log.Fatalf("connect: %v", err)
+ }
+ defer pool.Close()
+
+ if !*simulateOnly {
+ // Migrations first so the sandbox works on a fresh database.
+ if err := db.RunMigrations(cfg.DatabaseURL); err != nil {
+ log.Fatalf("migrations: %v", err)
+ }
+ seedCtx, cancel := context.WithTimeout(ctx, 120*time.Second)
+ if err := sandbox.Seed(seedCtx, pool, cfg); err != nil {
+ cancel()
+ log.Fatalf("seed: %v", err)
+ }
+ cancel()
+ }
+ if *seedOnly {
+ return
+ }
+
+ if err := sandbox.Simulate(ctx, pool, cfg); err != nil {
+ log.Fatalf("simulate: %v", err)
+ }
+}
diff --git a/deploy/config/env.example b/deploy/config/env.example
index 08b4de30..46a6a10c 100644
--- a/deploy/config/env.example
+++ b/deploy/config/env.example
@@ -43,6 +43,20 @@ SCHEMA_REGISTRY_URL=http://localhost:8081
SCHEMA_REGISTRY_KEY= # Optional - leave empty for local dev
SCHEMA_REGISTRY_SECRET= # Optional - leave empty for local dev
+# === Event codec ===
+# json is required wherever workers run: the worker command/result envelopes
+# carry untyped bodies that Avro cannot serialize. avro (the default when
+# unset) remains only for Schema Registry topics like tracking-events, which
+# keep their own Avro path regardless of this setting. Backend, consumer, and
+# every worker must share the same value.
+CODEC_PROVIDER=json
+
+# === Credential sealing ===
+# 64 hex chars (32 bytes). Seals mailbox SMTP/IMAP credentials at rest;
+# backend and consumer must share it. Losing it makes connected smtp_imap
+# mailboxes unusable until reconnected.
+CREDENTIALS_ENCRYPTION_KEY=
+
# === Backend API ===
API_HOST=0.0.0.0:8080
API_PORT=8080
diff --git a/docker-compose.yml b/docker-compose.yml
index fa2bebc3..57ab2011 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -41,6 +41,10 @@ x-worker-base: &worker-base
environment: &worker-env
<<: *aws-localstack
APP_ENV: dev
+ # Worker command/result envelopes carry `any` bodies; only the JSON
+ # codec serializes them. Backend + consumer must match.
+ CODEC_PROVIDER: json
+ MAIL_TLS_INSECURE: "true"
AWS_CONFIG_ENABLED: "false"
REDIS: redis://redis:6379
KAFKA_BOOTSTRAP_SERVERS: kafka:29092
@@ -179,6 +183,24 @@ services:
kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic "$$topic" --partitions 1 --replication-factor 1;
done
+ # One-shot: registers the tracking-events Avro schema. The Rust tracking
+ # service encodes fetch-only (no auto-register), so without this every
+ # open/click event fails serialization and is dropped. Mirrors
+ # TRACKING_EVENT_SCHEMA in tracking/src/kafka.rs.
+ schema-registry-init:
+ image: curlimages/curl:latest
+ depends_on:
+ schema-registry:
+ condition: service_started
+ entrypoint: ["/bin/sh", "-c"]
+ command:
+ - >-
+ until curl -sf http://schema-registry:8081/subjects >/dev/null; do sleep 2; done;
+ curl -sf -X POST -H 'Content-Type: application/vnd.schemaregistry.v1+json'
+ --data '{"schema":"{\"type\":\"record\",\"name\":\"TrackingEvent\",\"namespace\":\"com.warmbly.tracking\",\"fields\":[{\"name\":\"event_type\",\"type\":\"string\"},{\"name\":\"task_id\",\"type\":\"string\"},{\"name\":\"original_url\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"timestamp\",\"type\":\"string\"},{\"name\":\"user_agent\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"ip_hash\",\"type\":[\"null\",\"string\"],\"default\":null}]}"}'
+ http://schema-registry:8081/subjects/tracking-events-value/versions
+ && echo "tracking-events schema registered"
+
schema-registry:
image: confluentinc/cp-schema-registry:7.5.0
depends_on:
@@ -199,6 +221,11 @@ services:
- "-queue"
- "projects/local/locations/local/queues/default"
ports: ["8123:8123"]
+ # Native-dev callbacks: the emulator must reach the HOST-run backend at
+ # host.docker.internal (SELF_HOST default in the Makefile). Docker Desktop
+ # resolves it natively; this mapping makes it work on Linux too.
+ extra_hosts:
+ - "host.docker.internal:host-gateway"
mailpit:
image: axllent/mailpit:latest
@@ -207,12 +234,36 @@ services:
ports:
- "18025:8025" # web UI
- "11025:1025" # SMTP
+ # Accept SMTP AUTH with any credentials, without requiring TLS. The
+ # worker always authenticates, so the sandbox mailboxes (SMTP host ->
+ # mailpit) need the sink to accept their login (see /development/sandbox/).
+ environment:
+ MP_SMTP_AUTH_ACCEPT_ANY: 1
+ MP_SMTP_AUTH_ALLOW_INSECURE: 1
healthcheck:
test: ["CMD", "wget", "-q", "--spider", "http://localhost:8025"]
interval: 5s
timeout: 3s
retries: 10
+ # Sandbox mailbox host: a real IMAP server so the worker's genuine
+ # IMAP sync path runs locally (mailpit is SMTP-only). Rootless Dovecot
+ # 2.4 with a static passdb: ANY username logs in with the single
+ # USER_PASSWORD below, and the maildir is auto-created on first use.
+ # Used by `make sandbox` (see docs /development/sandbox/).
+ dovecot:
+ image: dovecot/dovecot:latest
+ environment:
+ USER_PASSWORD: "{PLAIN}sandbox"
+ ports:
+ - "10143:31143" # IMAP (cleartext auth disabled; debugging only)
+ - "10993:31993" # IMAPS (self-signed cert; MAIL_TLS_INSECURE=true clients)
+ healthcheck:
+ test: ["CMD", "doveadm", "service", "status", "imap-login"]
+ interval: 5s
+ timeout: 3s
+ retries: 10
+
localstack:
image: localstack/localstack:3.7
ports:
@@ -262,6 +313,8 @@ services:
environment:
<<: *aws-localstack
APP_ENV: dev
+ CODEC_PROVIDER: json
+ CREDENTIALS_ENCRYPTION_KEY: 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
AWS_CONFIG_ENABLED: "false"
API_HOST: "0.0.0.0:8080"
@@ -348,6 +401,8 @@ services:
environment:
<<: *aws-localstack
APP_ENV: dev
+ CODEC_PROVIDER: json
+ CREDENTIALS_ENCRYPTION_KEY: 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
AWS_CONFIG_ENABLED: "false"
PRIMARY_DB: postgres://warmbly:warmbly@postgres:5432/warmbly_dev?sslmode=disable
diff --git a/docs/content/docs/development/local-development.mdx b/docs/content/docs/development/local-development.mdx
index 23139d40..0e4fe2b5 100644
--- a/docs/content/docs/development/local-development.mdx
+++ b/docs/content/docs/development/local-development.mdx
@@ -23,10 +23,11 @@ For native development (running services outside Docker against the containerize
Day-one:
```bash
-make infra # postgres, redis, kafka, mailpit, localstack, etc. (leave running)
+make infra # postgres, redis, kafka, mailpit, dovecot, localstack, etc. (leave running)
make app # backend, consumer, worker, tracking, realtime, web (hot reload)
make sim # adds premium + dedicated workers; full simulation (prod-image flow)
make seed # load rich fixtures (3 orgs, 6 mailboxes, a campaign)
+make sandbox # fully working demo environment (see /development/sandbox/)
make tools # debugging UIs (kafka-ui at :18090)
make reset # nuke everything including volumes, start over
```
@@ -80,6 +81,7 @@ After `make infra && make app`:
- **localstack**: KMS + S3 emulation
- **stripe-mock**: Stripe API surrogate
- **mailpit**: SMTP catcher with a web UI
+- **dovecot**: real IMAP server hosting sandbox mailbox inboxes
- **cloud-tasks-emulator**: Google Cloud Tasks surrogate
- **backend**, **consumer**, **tracking**, **realtime**, **web**: app services
- **worker-shared-1**: one worker bound to the shared profile
@@ -97,6 +99,7 @@ Mostly standard ports. A few are offset because their defaults conflict too ofte
| Realtime | `http://localhost:4000` |
| Web (Vite dev) | `http://localhost:5173` |
| Mailpit | `http://localhost:18025` |
+| Dovecot IMAP / IMAPS | `localhost:10143` / `localhost:10993` |
| Kafka | `localhost:9092` |
| Schema Registry | `http://localhost:8081` |
| Postgres | `localhost:15432` |
@@ -205,13 +208,13 @@ mix deps.get && mix phx.server
## Mailpit
-All outbound mail is captured by Mailpit. The backend uses plain SMTP (`mailpit:1025`) in dev rather than SES, so no AWS credentials are needed.
+All outbound mail is captured by Mailpit. The backend uses plain SMTP (`mailpit:1025`) in dev rather than SES, so no AWS credentials are needed. Mailpit accepts SMTP AUTH with any credentials (no TLS required), so worker sends from sandbox mailboxes are captured too.
- Web UI: `http://localhost:18025`
- SMTP from inside docker: `mailpit:1025`
- SMTP from host: `localhost:11025`
-Note: Mailpit speaks SMTP only, not IMAP. The worker's IMAP sync path is not exercised by the default stack; for that, add a real IMAP server (e.g. GreenMail) to the stack.
+Note: Mailpit speaks SMTP only, not IMAP. The worker's IMAP sync path is exercised through the Dovecot service instead; see the [sandbox](/development/sandbox/) for the full live-mail environment.
## Email templates
@@ -279,8 +282,11 @@ curl http://localhost:8081/subjects/tracking-events-value/versions/latest | jq
**LocalStack init failed.** Check `docker compose logs localstack-init`. Usually means LocalStack itself isn't ready yet; the dependency wait should handle it, but re-running `make infra` works.
+**Scheduled sends never fire (native flow).** The cloud-tasks emulator runs in docker and calls the natively-run backend back at `host.docker.internal:8080` (the `SELF_HOST` default). Check `docker compose logs cloud-tasks-emulator` for connection errors; the task reconcilers cancel and re-enqueue lost callbacks within a few minutes either way.
+
## Next steps
+- [Sandbox](/development/sandbox/): a fully working demo environment on this stack
- [Architecture](/development/architecture/): control vs execution plane, encryption model
- [Deployment guide](/development/deployment-guide/): running in production
- [Event system](/development/events/): Kafka event reference
diff --git a/docs/content/docs/development/meta.json b/docs/content/docs/development/meta.json
index 3d276aaf..dc891a62 100644
--- a/docs/content/docs/development/meta.json
+++ b/docs/content/docs/development/meta.json
@@ -6,6 +6,7 @@
"pages": [
"architecture",
"local-development",
+ "sandbox",
"deployment-guide",
"events"
]
diff --git a/docs/content/docs/development/sandbox.mdx b/docs/content/docs/development/sandbox.mdx
new file mode 100644
index 00000000..2e4e50e2
--- /dev/null
+++ b/docs/content/docs/development/sandbox.mdx
@@ -0,0 +1,103 @@
+---
+title: Sandbox
+description: A fully working local demo environment, with live sending, replies, opens, clicks, and warmup running end to end.
+icon: FlaskConical
+---
+
+The sandbox turns a local stack into a living product demo. It seeds a paid showcase organization ("Sunrise Labs") whose mailboxes really send and really sync, then runs a simulator that plays the internet: it delivers captured mail into recipient inboxes, opens tracking pixels, clicks tracked links, and replies as the seeded contacts. Campaign stats climb, the unibox fills with threaded replies, warmup interactions verify, and the dashboard updates in realtime, all through production code paths. The only simulated thing is the humans.
+
+## Quick start
+
+Run each in its own terminal:
+
+```bash
+make infra # postgres, kafka, mailpit, dovecot, localstack, ... (once)
+make run # backend + consumer + both shared workers
+make tracking # open/click tracking (Rust; needs cargo + cmake)
+make realtime # websocket fanout (Elixir; optional but recommended)
+make web # dashboard on http://localhost:5173
+make sandbox # seed the showcase org + run the simulator (foreground)
+```
+
+Then sign in at `http://localhost:5173`:
+
+| Field | Value |
+|-------|-------|
+| Email | `sandbox@warmbly.test` |
+| Password | `password123` |
+| Organization | Sunrise Labs (Starter plan) |
+
+Campaign sending starts within about five minutes (the campaign reconciler's first sweep), warmup within about ten. From there the simulator keeps everything moving: watch `make sandbox`'s output for a running feed of `delivered` / `opened` / `clicked` / `replied` lines, and Mailpit at `http://localhost:18025` for every message on the wire.
+
+## What gets seeded
+
+`make sandbox` (or `make sandbox-seed` for seeding without the simulator) is idempotent and safe to re-run. It provisions:
+
+- the full `internal/seed` fixture first (plans, workers, the demo orgs), so it works on a fresh database
+- the Sunrise Labs org on an active Starter subscription, owned by `sandbox@warmbly.test`
+- 6 sender mailboxes on `@sunrise.test`, assigned to the premium shared worker (paid orgs place onto premium workers; `make run` now runs both native workers), warmup enabled 10 days ago (mid-ramp), premium warmup pool, demo-friendly pacing (90s minimum gap, 100/day cap)
+- two active campaigns with tracking enabled ("Sunrise Q3 launch outreach", 3 steps, 24 contacts; "Agency partnerships", 2 steps, 12 contacts) plus one draft; two contacts are unsubscribed so suppression is visible
+- working credentials for every `smtp_imap` account in the database, including the older `make seed` fixtures, sealed with `CREDENTIALS_ENCRYPTION_KEY` so the worker can decrypt and use them
+
+## How it works
+
+Real mail servers stand in for the internet's mail infrastructure:
+
+- **Mailpit** is the SMTP sink. Every mailbox's outbound server points at it, so all sends (campaign and warmup) are captured and visible in its UI.
+- **Dovecot** hosts the mailboxes' inboxes. It accepts any username with the password `sandbox` and auto-creates the maildir on first use. The worker's real IMAP sync polls it every minute.
+
+The simulator bridges the two and plays every human:
+
+1. It polls Mailpit's API for new messages (the read flag is its cursor).
+2. Mail addressed to a hosted mailbox (for example warmup mail between pool members) is appended into that user's Dovecot inbox. The recipient's worker sync picks it up, the warmup token verifies, and engagement actions run against the real IMAP folders.
+3. Mail addressed to a seeded contact triggers a persona: after humanized delays, most contacts open the pixel against the local tracking service, some click the tracked link, and some reply. Replies are composed with correct `In-Reply-To` threading and appended into the sending mailbox's inbox, so reply attribution, the reply classifier, `EMAIL_REPLIED` realtime events, and the unibox all behave exactly as in production. A slice of replies are out-of-office autoresponders (with `Auto-Submitted` headers) so classifier gating is visible too.
+
+Personas are derived from a hash of the contact's address, so behavior is stable across restarts without any extra state: roughly 85% open, 43% of openers click, 36% of openers reply, with reply tones split across interested, question, not-interested, and out-of-office.
+
+## Pieces involved
+
+| Piece | Role |
+|-------|------|
+| `cmd/sandbox` | seeder + simulator binary (`-seed-only`, `-simulate-only`) |
+| `internal/sandbox` | seeding, Mailpit client, personas, IMAP delivery |
+| `dovecot` compose service | IMAP host for sandbox inboxes (ports 10143/10993) |
+| `mailpit` compose service | SMTP sink + capture API (ports 11025/18025) |
+| `CREDENTIALS_ENCRYPTION_KEY` | seals mailbox credentials at rest; fixed dev value in the Makefile |
+| `MAIL_TLS_INSECURE=true` | worker-only dev knob: skip TLS verification and allow the TLS-less Mailpit; never set in production |
+
+## Environment variables
+
+`make sandbox` sets everything for the default stack. When running `cmd/sandbox` by hand:
+
+| Variable | Default | Meaning |
+|----------|---------|---------|
+| `PRIMARY_DB` | dev postgres on `:15432` | database for seeding and directory lookups |
+| `MAILPIT_URL` | `http://localhost:18025` | Mailpit API base |
+| `TRACKING_URL` | `http://localhost:3000` | tracking service for pixel/click hits |
+| `DOVECOT_IMAP_ADDR` | `localhost:10993` | where the simulator appends mail |
+| `DOVECOT_PASSWORD` | `sandbox` | Dovecot's static password |
+| `SANDBOX_SMTP_HOST` / `SANDBOX_SMTP_PORT` | `localhost` / `11025` | seeded as the mailboxes' outbound server (the worker dials it) |
+| `SANDBOX_IMAP_HOST` / `SANDBOX_IMAP_PORT` | `localhost` / `10993` | seeded as the mailboxes' inbound server (the worker dials it) |
+| `CREDENTIALS_ENCRYPTION_KEY` | set by the Makefile | must match the backend's key |
+
+## Demo script
+
+A tour that shows the platform working end to end:
+
+1. **Campaigns**: open the Sunrise Q3 launch campaign. Sent counts tick up as the scheduler paces sends across the six mailboxes; opens, clicks, and replies accumulate within minutes of each send.
+2. **Unibox**: replies arrive threaded under the original message with full bodies, synced by the worker over IMAP. Out-of-office autoresponders appear but do not count as human replies.
+3. **Mailboxes**: each sender shows warmup activity climbing its ramp; the warmup pool exchanges verified mail between accounts.
+4. **Mailpit** (`http://localhost:18025`): the raw wire view; every message the platform sends, including warmup tokens and tracking rewrites.
+5. **Realtime**: with `make realtime` and a second browser window, watch counts and inbox rows update live without refreshing.
+
+## Troubleshooting
+
+**No sends after ~5 minutes.** Check the backend log for campaign reconciler lines and that the worker is running with the account loaded (`email account added to worker`). The worker learns about mailboxes from the backend's reconciler within a minute of both being up.
+
+**Sends fail with TLS errors.** The worker must run with `MAIL_TLS_INSECURE=true` (set by `make worker`). Production configurations must never set it.
+
+**Replies missing from the unibox.** Confirm the simulator is running and `make consumer` is up: delivery is simulator → Dovecot → worker IMAP sync → Kafka → consumer.
+
+**Opens/clicks not counting.** `make tracking` must be running (it needs `cargo` and `cmake` on the host), and the consumer processes `tracking-events` into campaign stats.
+
+**Start over.** `make db-wipe && make migrate && make sandbox` reseeds from scratch; Dovecot and Mailpit state can be cleared with `docker compose -p warmbly rm -sf dovecot mailpit` followed by `make infra`.
diff --git a/internal/api/handler/admin_workers_ssh.go b/internal/api/handler/admin_workers_ssh.go
index a46df85d..eaf60c39 100644
--- a/internal/api/handler/admin_workers_ssh.go
+++ b/internal/api/handler/admin_workers_ssh.go
@@ -406,7 +406,7 @@ func (h *Handler) AdminRebootWorker(c *gin.Context) {
// Body:
//
// {
-// "user_id": "uuid", // the org/user that gets exclusive use
+// "organization_id": "uuid", // the org that gets exclusive use
// "subscription_id": "uuid", // their active sub
// "drain_to_worker_id": "uuid|null" // optional: target for evicted accounts.
// // null = let assignment service pick
@@ -419,7 +419,7 @@ func (h *Handler) AdminRebootWorker(c *gin.Context) {
// are individually idempotent: re-running the endpoint with the same
// inputs converges.
type convertToDedicatedBody struct {
- UserID string `json:"user_id" binding:"required"`
+ OrganizationID string `json:"organization_id" binding:"required"`
SubscriptionID string `json:"subscription_id" binding:"required"`
DrainToWorkerID *string `json:"drain_to_worker_id"`
}
@@ -435,9 +435,9 @@ func (h *Handler) AdminConvertWorkerToDedicated(c *gin.Context) {
errx.JSON(c, errx.New(errx.BadRequest, "invalid request body"))
return
}
- userID, err := uuid.Parse(body.UserID)
+ orgID, err := uuid.Parse(body.OrganizationID)
if err != nil {
- errx.JSON(c, errx.New(errx.BadRequest, "invalid user_id"))
+ errx.JSON(c, errx.New(errx.BadRequest, "invalid organization_id"))
return
}
subID, err := uuid.Parse(body.SubscriptionID)
@@ -506,7 +506,7 @@ func (h *Handler) AdminConvertWorkerToDedicated(c *gin.Context) {
created, err := h.WorkerRepo.CreateDedicatedAssignmentIfNotExists(c.Request.Context(), &models.DedicatedWorkerAssignment{
ID: uuid.New(),
WorkerID: id,
- UserID: userID,
+ OrganizationID: orgID,
SubscriptionID: subID,
AssignedAt: time.Now(),
})
@@ -516,7 +516,7 @@ func (h *Handler) AdminConvertWorkerToDedicated(c *gin.Context) {
}
h.audit(c, "convert_to_dedicated", models.AuditEntityWorker, &id, map[string]string{
- "user_id": userID.String(),
+ "organization_id": orgID.String(),
"subscription_id": subID.String(),
"drained_to": movedTo,
"accounts_moved": itoa(len(accountIDs)),
diff --git a/internal/api/handler/webhook_tasks_email.go b/internal/api/handler/webhook_tasks_email.go
index a73b87bf..fbd415b7 100644
--- a/internal/api/handler/webhook_tasks_email.go
+++ b/internal/api/handler/webhook_tasks_email.go
@@ -23,7 +23,9 @@ func (h *Handler) HandleEmailTask(c *gin.Context) {
return
}
- if err := h.TasksService.HandleEmailTask(&taskPayload); err != nil {
+ // Dispatch by the task row's type: every enqueue targets this one webhook
+ // URL, so campaign and user-email callbacks land here too.
+ if err := h.TasksService.HandleTask(&taskPayload); err != nil {
errx.Handle(c, err)
return
}
diff --git a/internal/app/cipher/cipher.go b/internal/app/cipher/cipher.go
index 021e2c35..565ffa2b 100644
--- a/internal/app/cipher/cipher.go
+++ b/internal/app/cipher/cipher.go
@@ -12,11 +12,15 @@ type Cipher struct {
}
func (s *cipherService) Cipher(ctx context.Context, orgID uuid.UUID) (*Cipher, error) {
- key, err := s.getDecryptedKey(ctx, orgID)
- if err != nil {
- return nil, err
+ // Cache hit: reuse the decrypted DEK. Any miss or cache error (redis.Nil
+ // on first use of an org's key) falls through to the KMS path — a cache
+ // problem must never block crypto.
+ if key, err := s.getDecryptedKey(ctx, orgID); err == nil && len(key) > 0 {
+ return &Cipher{plainDEK: key}, nil
}
+ var key []byte
+
encDEKB64, err := s.encryptedKeys.Get(ctx, orgID)
if err != nil {
return nil, err
diff --git a/internal/app/consumer/events.go b/internal/app/consumer/events.go
index 17b985d6..e3d8a09b 100644
--- a/internal/app/consumer/events.go
+++ b/internal/app/consumer/events.go
@@ -2,6 +2,7 @@ package jobs
import (
"context"
+ "encoding/json"
"errors"
"fmt"
@@ -44,9 +45,18 @@ func (w *JobsService) InitEvents() {
func Register[T any](w *JobsService, eventType models.JobEventType, handler EventHandler[T]) {
w.eventHandlers[eventType] = func(ctx context.Context, body any) error {
- data, ok := body.(T)
- if !ok {
- return fmt.Errorf("invalid event body for type %v", eventType)
+ if data, ok := body.(T); ok {
+ return handler(ctx, data)
+ }
+ // The JSON codec decodes the envelope's `body` into map[string]any;
+ // round-trip it into the typed payload.
+ raw, err := json.Marshal(body)
+ if err != nil {
+ return fmt.Errorf("invalid event body for type %v: %w", eventType, err)
+ }
+ var data T
+ if err := json.Unmarshal(raw, &data); err != nil {
+ return fmt.Errorf("invalid event body for type %v: %w", eventType, err)
}
return handler(ctx, data)
}
diff --git a/internal/app/consumer/receive.go b/internal/app/consumer/receive.go
index 51cdf0bd..8d47eb7b 100644
--- a/internal/app/consumer/receive.go
+++ b/internal/app/consumer/receive.go
@@ -11,12 +11,16 @@ import (
func (s *JobsService) Receive(msg *cfk.Message) error {
var event models.JobEvent
- if err := s.Consumer.Avrov2.Deser.DeserializeInto(*msg.TopicPartition.Topic, msg.Value, &event); err != nil {
- return err
- }
-
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
+ if s.Codec != nil {
+ if err := s.Codec.Deserialize(ctx, *msg.TopicPartition.Topic, msg.Value, &event); err != nil {
+ return err
+ }
+ } else if err := s.Consumer.Avrov2.Deser.DeserializeInto(*msg.TopicPartition.Topic, msg.Value, &event); err != nil {
+ return err
+ }
+
return s.HandleEvent(ctx, &event)
}
diff --git a/internal/app/consumer/service.go b/internal/app/consumer/service.go
index 443dc3d5..e9f496bd 100644
--- a/internal/app/consumer/service.go
+++ b/internal/app/consumer/service.go
@@ -8,6 +8,7 @@ import (
workerapp "github.com/warmbly/warmbly/internal/app/worker"
"github.com/warmbly/warmbly/internal/events"
"github.com/warmbly/warmbly/internal/infrastructure/cache"
+ "github.com/warmbly/warmbly/internal/infrastructure/codec"
"github.com/warmbly/warmbly/internal/infrastructure/kafka"
"github.com/warmbly/warmbly/internal/infrastructure/pubsub"
"github.com/warmbly/warmbly/internal/models"
@@ -15,7 +16,11 @@ import (
)
type JobsService struct {
- Consumer *kafka.Consumer
+ Consumer *kafka.Consumer
+ // Codec decodes bus payloads (jobs.worker-events); it must match the
+ // CODEC_PROVIDER the producing services run with. When nil, the
+ // consumer's attached Avro deserializer is used.
+ Codec codec.Codec
UniboxRepository repository.UniboxRepository
MailboxRepository repository.MailboxRepository
EmailRepository repository.EmailRepository
diff --git a/internal/app/worker/assignment.go b/internal/app/worker/assignment.go
index 744c2ef8..e13dc052 100644
--- a/internal/app/worker/assignment.go
+++ b/internal/app/worker/assignment.go
@@ -96,7 +96,7 @@ func (s *workerAssignmentService) AssignWorkerToEmail(ctx context.Context, email
}
if plan != nil && plan.DedicatedWorkers > 0 {
// Check if org has a dedicated worker
- dedicatedWorker, err := s.workerRepo.GetDedicatedWorkerByUserID(ctx, orgID)
+ dedicatedWorker, err := s.workerRepo.GetDedicatedWorkerByOrgID(ctx, orgID)
if err != nil {
return nil, err
}
@@ -114,7 +114,7 @@ func (s *workerAssignmentService) AssignWorkerToEmail(ctx context.Context, email
return nil, aerr
}
if aerr == nil || errors.Is(aerr, ErrOrgAlreadyAssigned) {
- dedicatedWorker, err = s.workerRepo.GetDedicatedWorkerByUserID(ctx, orgID)
+ dedicatedWorker, err = s.workerRepo.GetDedicatedWorkerByOrgID(ctx, orgID)
if err != nil {
return nil, err
}
@@ -426,7 +426,7 @@ func (s *workerAssignmentService) AssignDedicatedWorker(ctx context.Context, org
assignment := &models.DedicatedWorkerAssignment{
ID: uuid.New(),
WorkerID: worker.ID,
- UserID: orgID,
+ OrganizationID: orgID,
SubscriptionID: subscriptionID,
AssignedAt: time.Now(),
}
@@ -465,7 +465,7 @@ func (s *workerAssignmentService) ReleaseDedicatedWorker(ctx context.Context, or
// GetDedicatedWorker gets the dedicated worker for an organization
func (s *workerAssignmentService) GetDedicatedWorker(ctx context.Context, orgID uuid.UUID) (*models.Worker, error) {
- return s.workerRepo.GetDedicatedWorkerByUserID(ctx, orgID)
+ return s.workerRepo.GetDedicatedWorkerByOrgID(ctx, orgID)
}
// MigrateOrgToPremiumWorkers migrates all org's emails from free to premium workers
@@ -552,7 +552,7 @@ func (s *workerAssignmentService) MigrateOrgToDedicated(ctx context.Context, org
}
// Get the dedicated worker
- dedicatedWorker, err := s.workerRepo.GetDedicatedWorkerByUserID(ctx, orgID)
+ dedicatedWorker, err := s.workerRepo.GetDedicatedWorkerByOrgID(ctx, orgID)
if err != nil {
return err
}
diff --git a/internal/app/worker/assignment_test.go b/internal/app/worker/assignment_test.go
index e83dc534..321d7d0a 100644
--- a/internal/app/worker/assignment_test.go
+++ b/internal/app/worker/assignment_test.go
@@ -51,7 +51,7 @@ type stubWorkerRepo struct {
promotedToPool *models.Worker // PromoteWorkerToPool result
}
-func (r *stubWorkerRepo) GetDedicatedWorkerByUserID(_ context.Context, _ uuid.UUID) (*models.Worker, error) {
+func (r *stubWorkerRepo) GetDedicatedWorkerByOrgID(_ context.Context, _ uuid.UUID) (*models.Worker, error) {
return r.dedicatedForOrg, nil
}
func (r *stubWorkerRepo) GetSharedWorkersByTier(_ context.Context, freeTier bool) ([]models.Worker, error) {
diff --git a/internal/app/worker/event_email_validation.go b/internal/app/worker/event_email_validation.go
index a8e43e4f..44b068eb 100644
--- a/internal/app/worker/event_email_validation.go
+++ b/internal/app/worker/event_email_validation.go
@@ -7,26 +7,13 @@ import (
"github.com/getsentry/sentry-go"
"github.com/warmbly/warmbly/internal/email"
- "github.com/warmbly/warmbly/internal/errx"
"github.com/warmbly/warmbly/internal/models"
)
-func (w *WorkerService) HandleEmailValidation(ctx context.Context, body any) error {
+func (w *WorkerService) HandleEmailValidation(ctx context.Context, data models.EventWorkerEmailValidation) error {
ctx, cancel := context.WithDeadline(ctx, time.Now().Add(5*time.Second))
defer cancel()
- data, ok := body.(models.EventWorkerEmailValidation)
- if !ok {
- err := errx.ErrInvalidEventFormat
- sentry.WithScope(func(scope *sentry.Scope) {
- scope.SetTag("event_type", string(models.WorkerEventTypeEmailValidation))
- scope.SetTag("process_id", data.ProcessID.String())
- scope.SetTag("org_id", data.OrgID.String())
- sentry.CaptureException(err)
- })
- return err
- }
-
cipher, err := w.CipherService.Cipher(ctx, data.OrgID)
if err != nil {
sentry.CaptureException(err)
diff --git a/internal/app/worker/event_send_email.go b/internal/app/worker/event_send_email.go
index 9e20c44d..d16def42 100644
--- a/internal/app/worker/event_send_email.go
+++ b/internal/app/worker/event_send_email.go
@@ -15,13 +15,7 @@ import (
"github.com/warmbly/warmbly/internal/pkg/emsg"
)
-func (w *WorkerService) HandleSendEmail(ctx context.Context, body any) error {
- sendEmail, ok := body.(models.SendEmail)
- if !ok {
- log.Debug().Msg("Invalid HandleSendEmail body type")
- return fmt.Errorf("invalid body type")
- }
-
+func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.SendEmail) error {
log.Info().
Str("task_id", sendEmail.TaskID.String()).
Str("email_id", sendEmail.EmailID.String()).
diff --git a/internal/app/worker/event_warmup_action.go b/internal/app/worker/event_warmup_action.go
index c37f770e..1f8899f7 100644
--- a/internal/app/worker/event_warmup_action.go
+++ b/internal/app/worker/event_warmup_action.go
@@ -2,7 +2,6 @@ package worker
import (
"context"
- "fmt"
"github.com/rs/zerolog/log"
"github.com/warmbly/warmbly/internal/app/worker/wmail"
@@ -19,13 +18,7 @@ import (
// the delayed leg (read / important / star) when its fire_at passes, each with
// DelaySeconds=0. That makes the dwell survive a worker restart, which the old
// in-process time.AfterFunc here could not.
-func (w *WorkerService) HandleWarmupAction(ctx context.Context, body any) error {
- action, ok := body.(models.WarmupEmailAction)
- if !ok {
- log.Debug().Msg("Invalid HandleWarmupAction body type")
- return fmt.Errorf("invalid body type")
- }
-
+func (w *WorkerService) HandleWarmupAction(ctx context.Context, action models.WarmupEmailAction) error {
log.Info().
Str("email_id", action.EmailID.String()).
Str("gmail_id", action.GmailID).
diff --git a/internal/app/worker/events.go b/internal/app/worker/events.go
index 0d24c280..20d16522 100644
--- a/internal/app/worker/events.go
+++ b/internal/app/worker/events.go
@@ -2,6 +2,7 @@ package worker
import (
"context"
+ "encoding/json"
"errors"
"fmt"
@@ -29,9 +30,18 @@ func (w *WorkerService) InitEvents() {
func Register[T any](w *WorkerService, eventType models.WorkerEventType, handler EventHandler[T]) {
w.eventHandlers[eventType] = func(ctx context.Context, body any) error {
- data, ok := body.(T)
- if !ok {
- return fmt.Errorf("invalid event body for type %v", eventType)
+ if data, ok := body.(T); ok {
+ return handler(ctx, data)
+ }
+ // The JSON codec decodes the envelope's `body` into map[string]any;
+ // round-trip it into the typed payload.
+ raw, err := json.Marshal(body)
+ if err != nil {
+ return fmt.Errorf("invalid event body for type %v: %w", eventType, err)
+ }
+ var data T
+ if err := json.Unmarshal(raw, &data); err != nil {
+ return fmt.Errorf("invalid event body for type %v: %w", eventType, err)
}
return handler(ctx, data)
}
diff --git a/internal/app/worker/wmail/start_sync.go b/internal/app/worker/wmail/start_sync.go
index f8abc157..9e2b8e6e 100644
--- a/internal/app/worker/wmail/start_sync.go
+++ b/internal/app/worker/wmail/start_sync.go
@@ -2,6 +2,7 @@ package wmail
import (
"context"
+ "fmt"
"time"
"github.com/rs/zerolog/log"
@@ -21,9 +22,7 @@ func (w *WMail) StartSyncWorker(ctx context.Context) {
}
// Run an initial sync immediately so the inbox is fresh on startup.
- if err := w.SyncMail(ctx); err != nil {
- log.Warn().Err(err).Str("email_id", w.ID.String()).Msg("initial mail sync failed")
- }
+ w.syncOnce(ctx)
ticker := time.NewTicker(interval)
defer ticker.Stop()
@@ -33,10 +32,24 @@ func (w *WMail) StartSyncWorker(ctx context.Context) {
case <-ctx.Done():
return
case <-ticker.C:
- if err := w.SyncMail(ctx); err != nil {
- w.CaptureError(err)
- log.Warn().Err(err).Str("email_id", w.ID.String()).Msg("mail sync error")
- }
+ w.syncOnce(ctx)
}
}
}
+
+// syncOnce runs one sync pass, containing panics: the worker is multi-tenant,
+// so one mailbox's bad server response must not take down every other
+// account's sync and send loops.
+func (w *WMail) syncOnce(ctx context.Context) {
+ defer func() {
+ if r := recover(); r != nil {
+ err := fmt.Errorf("mail sync panic: %v", r)
+ w.CaptureError(err)
+ log.Error().Err(err).Str("email_id", w.ID.String()).Msg("mail sync panicked")
+ }
+ }()
+ if err := w.SyncMail(ctx); err != nil {
+ w.CaptureError(err)
+ log.Warn().Err(err).Str("email_id", w.ID.String()).Msg("mail sync error")
+ }
+}
diff --git a/internal/app/worker/wmail/sync_imap.go b/internal/app/worker/wmail/sync_imap.go
index df6f6ec2..31ae71af 100644
--- a/internal/app/worker/wmail/sync_imap.go
+++ b/internal/app/worker/wmail/sync_imap.go
@@ -25,9 +25,18 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError {
return nil
}
- if err := w.SmtpImapData.ImapClient.FetchChanges(ctx, 0); err != nil {
+ // FETCH requires a selected mailbox; the select also arms
+ // CONDSTORE for the ChangedSince filtering. An empty mailbox is
+ // skipped: 1:* on zero messages is a server error.
+ count, err := w.SmtpImapData.ImapClient.SelectForSync(box.Name)
+ if err != nil {
return err
}
+ if count > 0 {
+ if err := w.SmtpImapData.ImapClient.FetchChanges(ctx, 0); err != nil {
+ return err
+ }
+ }
w.SmtpImapData.Mailboxes = append(w.SmtpImapData.Mailboxes, &box)
continue
@@ -35,9 +44,15 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError {
if befBox.HighestModSeq != box.HighestModSeq {
w.SmtpImapData.mailbox = box.UIDValidity
- if err := w.SmtpImapData.ImapClient.FetchChanges(ctx, befBox.HighestModSeq); err != nil {
+ count, err := w.SmtpImapData.ImapClient.SelectForSync(box.Name)
+ if err != nil {
return err
}
+ if count > 0 {
+ if err := w.SmtpImapData.ImapClient.FetchChanges(ctx, befBox.HighestModSeq); err != nil {
+ return err
+ }
+ }
}
if befBox.HighestModSeq != box.HighestModSeq || befBox.Name != box.Name || !slices.Equal(befBox.Attrs, box.Attrs) {
diff --git a/internal/client/netbind/netbind.go b/internal/client/netbind/netbind.go
index c7987616..80f3b583 100644
--- a/internal/client/netbind/netbind.go
+++ b/internal/client/netbind/netbind.go
@@ -18,6 +18,7 @@ import (
const (
envBindIP = "WORKER_BIND_IP"
+ envInsecureTLS = "MAIL_TLS_INSECURE"
defaultTimeout = 10 * time.Second
)
@@ -57,8 +58,26 @@ func Dialer(local *net.TCPAddr) *net.Dialer {
// TLSDialer mirrors Dialer for TLS connections (IMAP, HTTPS).
func TLSDialer(local *net.TCPAddr, cfg *tls.Config) *tls.Dialer {
+ if cfg != nil && InsecureTLS() {
+ cfg.InsecureSkipVerify = true
+ }
return &tls.Dialer{
NetDialer: Dialer(local),
Config: cfg,
}
}
+
+var (
+ insecureOnce sync.Once
+ insecureTLS bool
+)
+
+// InsecureTLS reports whether MAIL_TLS_INSECURE=true is set. Local-dev-only
+// escape hatch for the sandbox stack (mailpit/dovecot with self-signed or no
+// TLS); production deployments never set it, so mail TLS stays verified.
+func InsecureTLS() bool {
+ insecureOnce.Do(func() {
+ insecureTLS = os.Getenv(envInsecureTLS) == "true"
+ })
+ return insecureTLS
+}
diff --git a/internal/client/smtpimap/imap/client.go b/internal/client/smtpimap/imap/client.go
index 81daa477..411e4fa0 100644
--- a/internal/client/smtpimap/imap/client.go
+++ b/internal/client/smtpimap/imap/client.go
@@ -66,12 +66,7 @@ func (c *Client) Connect() *errx.MailError {
return errx.ErrMailServerUnreachable
}
- client := imapclient.New(conn, nil)
-
- caps := client.Caps()
- if !caps.Has(imap.CapCondStore) {
- return errx.ErrMailCondStoreNotSupported
- }
+ c.client = imapclient.New(conn, nil)
var xerr *errx.MailError
@@ -81,8 +76,18 @@ func (c *Client) Connect() *errx.MailError {
case models.AuthOAuth2:
xerr = c.oauth2Auth()
}
+ if xerr != nil {
+ return xerr
+ }
- return xerr
+ // CONDSTORE backs the ChangedSince incremental sync. Servers (Gmail,
+ // Dovecot, ...) typically advertise it only after authentication, so the
+ // check must run post-auth.
+ if !c.client.Caps().Has(imap.CapCondStore) {
+ return errx.ErrMailCondStoreNotSupported
+ }
+
+ return nil
}
func (c *Client) Close() error {
@@ -126,7 +131,14 @@ func (c *Client) oauth2Auth() *errx.MailError {
func (c *Client) Folders() ([]models.Mailbox, *errx.MailError) {
var resp []models.Mailbox
- cmd := c.client.List("", "%", nil)
+ // LIST-STATUS: without requesting these, f.Status is nil for every
+ // folder and the sync loop sees an empty account.
+ cmd := c.client.List("", "%", &imap.ListOptions{
+ ReturnStatus: &imap.StatusOptions{
+ UIDValidity: true,
+ HighestModSeq: true,
+ },
+ })
for f := cmd.Next(); f != nil; f = cmd.Next() {
if len(resp) >= config.MaxEmailFolders {
@@ -166,8 +178,25 @@ func (c *Client) Mailbox(mailbox string, uidvali, opts *imap.SelectOptions) erro
return nil
}
+// SelectForSync opens a mailbox read-only with CONDSTORE enabled and returns
+// its message count. FETCH is only valid against a selected mailbox, so the
+// sync loop must call this before FetchChanges; CONDSTORE on the SELECT is
+// what arms ChangedSince. The count lets the caller skip the fetch entirely
+// for an empty mailbox, where a 1:* set is a server error.
+func (c *Client) SelectForSync(mailbox string) (uint32, *errx.MailError) {
+ data, err := c.client.Select(mailbox, &imap.SelectOptions{ReadOnly: true, CondStore: true}).Wait()
+ if err != nil {
+ return 0, c.handleError(err)
+ }
+ return data.NumMessages, nil
+}
+
func (c *Client) FetchChanges(ctx context.Context, lastModSeq uint64) *errx.MailError {
- cmd := c.client.Fetch(&imap.SeqSet{}, &imap.FetchOptions{
+ // 1:* — an empty SeqSet has no encodable ranges and panics inside
+ // go-imap; ChangedSince narrows the result server-side.
+ var allMessages imap.SeqSet
+ allMessages.AddRange(1, 0)
+ cmd := c.client.Fetch(allMessages, &imap.FetchOptions{
UID: true,
Envelope: true,
BodyStructure: &imap.FetchItemBodyStructure{
diff --git a/internal/client/smtpimap/smtp/client.go b/internal/client/smtpimap/smtp/client.go
index 5f05377b..20025df2 100644
--- a/internal/client/smtpimap/smtp/client.go
+++ b/internal/client/smtpimap/smtp/client.go
@@ -228,9 +228,17 @@ func (c *Client) sendRaw(ctx context.Context, from string, to []string, data []b
defer client.Quit()
tlsConf := &tls.Config{
- ServerName: host,
+ ServerName: host,
+ InsecureSkipVerify: netbind.InsecureTLS(),
}
- if err := client.StartTLS(tlsConf); err != nil {
+ // TLS is mandatory. The MAIL_TLS_INSECURE dev knob additionally allows a
+ // server with no STARTTLS at all (the local mailpit sink) — never taken in
+ // production, where the env var is unset.
+ if ok, _ := client.Extension("STARTTLS"); ok {
+ if err := client.StartTLS(tlsConf); err != nil {
+ return errx.ErrMailServerUnreachable
+ }
+ } else if !netbind.InsecureTLS() {
return errx.ErrMailServerUnreachable
}
diff --git a/internal/infrastructure/db/migrations/000055_dedicated_worker_org_assignments.down.sql b/internal/infrastructure/db/migrations/000055_dedicated_worker_org_assignments.down.sql
new file mode 100644
index 00000000..1e5b8bb2
--- /dev/null
+++ b/internal/infrastructure/db/migrations/000055_dedicated_worker_org_assignments.down.sql
@@ -0,0 +1,23 @@
+ALTER TABLE dedicated_worker_assignments
+ DROP CONSTRAINT IF EXISTS dedicated_worker_assignments_organization_id_fkey;
+
+ALTER TABLE dedicated_worker_assignments
+ RENAME COLUMN organization_id TO user_id;
+
+-- Map org-keyed rows back to the org owner so the users FK can be restored.
+UPDATE dedicated_worker_assignments dwa
+SET user_id = o.owner_user_id
+FROM organizations o
+WHERE o.id = dwa.user_id;
+
+DELETE FROM dedicated_worker_assignments dwa
+WHERE NOT EXISTS (SELECT 1 FROM users u WHERE u.id = dwa.user_id);
+
+ALTER TABLE dedicated_worker_assignments
+ ADD CONSTRAINT dedicated_worker_assignments_user_id_fkey
+ FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE;
+
+ALTER TABLE dedicated_worker_assignments
+ RENAME CONSTRAINT unique_active_org_assignment TO unique_active_user_assignment;
+
+ALTER INDEX idx_dedicated_org RENAME TO idx_dedicated_user;
diff --git a/internal/infrastructure/db/migrations/000055_dedicated_worker_org_assignments.up.sql b/internal/infrastructure/db/migrations/000055_dedicated_worker_org_assignments.up.sql
new file mode 100644
index 00000000..c0d4a9d7
--- /dev/null
+++ b/internal/infrastructure/db/migrations/000055_dedicated_worker_org_assignments.up.sql
@@ -0,0 +1,32 @@
+-- Dedicated workers are organization assets, and the assignment service has
+-- always keyed dedicated_worker_assignments by organization id - but the
+-- column was named user_id with a users FK, so every runtime insert failed
+-- with an FK violation and paid dedicated-plan orgs could never bind a
+-- worker. Re-key the table to organizations.
+
+ALTER TABLE dedicated_worker_assignments
+ DROP CONSTRAINT IF EXISTS dedicated_worker_assignments_user_id_fkey;
+
+ALTER TABLE dedicated_worker_assignments
+ RENAME COLUMN user_id TO organization_id;
+
+-- Pre-rename rows (seed fixtures) hold owner user ids: remap each to that
+-- owner's organization, then drop anything unmappable - such rows were never
+-- reachable by the runtime lookups anyway.
+UPDATE dedicated_worker_assignments dwa
+SET organization_id = o.id
+FROM organizations o
+WHERE o.owner_user_id = dwa.organization_id
+ AND NOT EXISTS (SELECT 1 FROM organizations WHERE id = dwa.organization_id);
+
+DELETE FROM dedicated_worker_assignments dwa
+WHERE NOT EXISTS (SELECT 1 FROM organizations o WHERE o.id = dwa.organization_id);
+
+ALTER TABLE dedicated_worker_assignments
+ ADD CONSTRAINT dedicated_worker_assignments_organization_id_fkey
+ FOREIGN KEY (organization_id) REFERENCES organizations(id) ON DELETE CASCADE;
+
+ALTER TABLE dedicated_worker_assignments
+ RENAME CONSTRAINT unique_active_user_assignment TO unique_active_org_assignment;
+
+ALTER INDEX idx_dedicated_user RENAME TO idx_dedicated_org;
diff --git a/internal/infrastructure/storage/s3.go b/internal/infrastructure/storage/s3.go
index 1620bd0b..a888ecac 100644
--- a/internal/infrastructure/storage/s3.go
+++ b/internal/infrastructure/storage/s3.go
@@ -2,6 +2,7 @@ package storage
import (
"context"
+ "os"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -15,6 +16,14 @@ type Client struct {
func NewClient(ctx context.Context, cfg aws.Config, bucket string) (*Client, error) {
return &Client{
Bucket: bucket,
- Client: s3.NewFromConfig(cfg),
+ Client: s3.NewFromConfig(cfg, func(o *s3.Options) {
+ // Custom endpoints (LocalStack, MinIO) need path-style requests:
+ // virtual-hosted addressing puts the bucket in the hostname
+ // (bucket.localhost:4566), which these servers don't resolve as a
+ // bucket. Real AWS (no endpoint override) keeps the default.
+ if os.Getenv("AWS_ENDPOINT_URL") != "" || os.Getenv("AWS_ENDPOINT_URL_S3") != "" {
+ o.UsePathStyle = true
+ }
+ }),
}, nil
}
diff --git a/internal/models/worker.go b/internal/models/worker.go
index 9bcf505c..f26e788b 100644
--- a/internal/models/worker.go
+++ b/internal/models/worker.go
@@ -128,7 +128,7 @@ type UpdateWorker struct {
type DedicatedWorkerAssignment struct {
ID uuid.UUID `json:"id"`
WorkerID uuid.UUID `json:"worker_id"`
- UserID uuid.UUID `json:"user_id"`
+ OrganizationID uuid.UUID `json:"organization_id"`
SubscriptionID uuid.UUID `json:"subscription_id"`
AssignedAt time.Time `json:"assigned_at"`
ReleasedAt *time.Time `json:"released_at,omitempty"`
diff --git a/internal/pkg/encrypt/encrypter.go b/internal/pkg/encrypt/encrypter.go
index ace0aeb6..4c462912 100644
--- a/internal/pkg/encrypt/encrypter.go
+++ b/internal/pkg/encrypt/encrypter.go
@@ -7,6 +7,7 @@ import (
"encoding/hex"
"errors"
"io"
+ "os"
)
// Encrypted holds the hex-encoded ciphertext and nonce.
@@ -21,6 +22,17 @@ type Encrypter struct {
aead cipher.AEAD
}
+// FromEnv creates an Encrypter from the CREDENTIALS_ENCRYPTION_KEY env var
+// (64 hex chars = 32 bytes). Returns (nil, nil) when the var is unset so
+// callers can keep booting without credential sealing configured.
+func FromEnv() (*Encrypter, error) {
+ key := os.Getenv("CREDENTIALS_ENCRYPTION_KEY")
+ if key == "" {
+ return nil, nil
+ }
+ return NewEncrypterFromHex(key)
+}
+
// NewEncrypterFromHex creates an Encrypter from a 64-character hex key.
func NewEncrypterFromHex(hexKey string) (*Encrypter, error) {
key, err := hex.DecodeString(hexKey)
diff --git a/internal/repository/pg_admin.go b/internal/repository/pg_admin.go
index 98896579..9144e2f0 100644
--- a/internal/repository/pg_admin.go
+++ b/internal/repository/pg_admin.go
@@ -204,7 +204,7 @@ func (r *adminRepository) SearchUsers(ctx context.Context, search *models.AdminU
whereClause += ` AND EXISTS (SELECT 1 FROM user_bans ub WHERE ub.user_id = u.id)`
}
if search.HasDedicatedWorker {
- whereClause += ` AND EXISTS (SELECT 1 FROM dedicated_worker_assignments dwa WHERE dwa.user_id = u.id AND dwa.released_at IS NULL)`
+ whereClause += ` AND EXISTS (SELECT 1 FROM dedicated_worker_assignments dwa JOIN organizations o ON o.id = dwa.organization_id WHERE o.owner_user_id = u.id AND dwa.released_at IS NULL)`
}
// Count / numeric ranges
diff --git a/internal/repository/pg_email.go b/internal/repository/pg_email.go
index af02cb15..fae3b86a 100644
--- a/internal/repository/pg_email.go
+++ b/internal/repository/pg_email.go
@@ -94,12 +94,21 @@ type emailRepository struct {
Encrypt *encrypt.Encrypter
}
-func NewEmailRepostory(db *db.DB) EmailRepository {
+// NewEmailRepostory builds the email repository. enc seals SMTP/IMAP and
+// OAuth credentials at rest (CREDENTIALS_ENCRYPTION_KEY); nil is tolerated so
+// deployments without the key keep booting, but credential reads/writes then
+// fail with a captured error instead of a nil-pointer panic.
+func NewEmailRepostory(db *db.DB, enc *encrypt.Encrypter) EmailRepository {
return &emailRepository{
- DB: db,
+ DB: db,
+ Encrypt: enc,
}
}
+// errNoCredentialEncrypter is returned when credential sealing is attempted
+// without CREDENTIALS_ENCRYPTION_KEY configured.
+var errNoCredentialEncrypter = errors.New("credential encrypter not configured (set CREDENTIALS_ENCRYPTION_KEY)")
+
func (r *emailRepository) ExistsForUser(ctx context.Context, userID, email string) (bool, *errx.Error) {
var exists bool
query := `SELECT EXISTS(SELECT 1 FROM email_accounts WHERE user_id = $1 AND email = $2)`
@@ -292,6 +301,10 @@ func (r *emailRepository) NewOauthAccount(ctx context.Context, userID string, da
}
func (r *emailRepository) NewSMTPIMAPAccount(ctx context.Context, userID string, data models.NewSMTPIMAPAccount) (*models.Email, *errx.Error) {
+ if r.Encrypt == nil {
+ sentry.CaptureException(errNoCredentialEncrypter)
+ return nil, errx.InternalError()
+ }
tx, err := r.DB.Begin(ctx)
if err != nil {
db.CaptureError(err, "", nil, "begin")
@@ -931,7 +944,7 @@ func (r *emailRepository) SetWarmupLifecycle(ctx context.Context, userID, emailA
func (r *emailRepository) GetByID(ctx context.Context, emailAccountID uuid.UUID) (*models.Email, *errx.Error) {
query := `
SELECT
- ea.id, ea.user_id, ea.organization_id, ea.email, ea.name, ea.signature_plain, ea.signature_html, ea.signature_sync, ea.signature_code,
+ ea.id, ea.user_id, ea.organization_id, ea.worker_id, ea.email, ea.name, ea.signature_plain, ea.signature_html, ea.signature_sync, ea.signature_code,
ea.provider, ea.status, COALESCE(ea.last_synced_at, ea.created_at) AS last_synced_at, ea.last_id, ea.campaign_limit,
ea.min_wait_time, ea.reply_to, ea.tracking_domain, ea.tracking_domain_verified, ea.tracking_domain_verified_at, ea.warmup, ea.warmup_paused_at, ea.warmup_base,
ea.warmup_max, ea.warmup_increase, ea.warmup_reply_rate, ea.warmup_tag, ea.warmup_pool_type,
@@ -946,7 +959,7 @@ func (r *emailRepository) GetByID(ctx context.Context, emailAccountID uuid.UUID)
var i models.Email
err := r.DB.QueryRow(ctx, query, emailAccountID).Scan(
- &i.ID, &i.UserID, &i.OrganizationID, &i.Email, &i.Name, &i.SignaturePlain, &i.SignatureHTML, &i.SignatureSync, &i.SignatureCode,
+ &i.ID, &i.UserID, &i.OrganizationID, &i.WorkerID, &i.Email, &i.Name, &i.SignaturePlain, &i.SignatureHTML, &i.SignatureSync, &i.SignatureCode,
&i.Provider, &i.Status, &i.LastSyncedAt, &i.LastID, &i.CampaignLimit,
&i.MinWaitTime, &i.ReplyTo, &i.TrackingDomain, &i.TrackingDomainVerified, &i.TrackingDomainVerifiedAt, &i.Warmup, &i.WarmupPausedAt, &i.WarmupBase,
&i.WarmupMax, &i.WarmupIncrease, &i.WarmupReplyRate, &i.WarmupTag, &i.WarmupPoolType,
@@ -1132,6 +1145,10 @@ func (r *emailRepository) GetByCampaignSenders(ctx context.Context, userID strin
// GetSMTPCredentials retrieves SMTP/IMAP credentials for an email account
func (r *emailRepository) GetSMTPCredentials(ctx context.Context, emailAccountID uuid.UUID) (*SMTPCredentials, *errx.Error) {
+ if r.Encrypt == nil {
+ sentry.CaptureException(errNoCredentialEncrypter)
+ return nil, errx.InternalError()
+ }
query := `
SELECT smtp_host, smtp_port, smtp_user, smtp_password,
imap_host, imap_port, imap_user, imap_password
@@ -1192,6 +1209,10 @@ func (r *emailRepository) GetSMTPCredentials(ctx context.Context, emailAccountID
// GetOAuthCredentials retrieves OAuth credentials for an email account
func (r *emailRepository) GetOAuthCredentials(ctx context.Context, emailAccountID uuid.UUID) (*OAuthCredentials, *errx.Error) {
+ if r.Encrypt == nil {
+ sentry.CaptureException(errNoCredentialEncrypter)
+ return nil, errx.InternalError()
+ }
query := `
SELECT access_token, refresh_token, expires_at
FROM email_accounts_oauth
diff --git a/internal/repository/pg_task.go b/internal/repository/pg_task.go
index 285eeed1..d027932e 100644
--- a/internal/repository/pg_task.go
+++ b/internal/repository/pg_task.go
@@ -107,6 +107,13 @@ type TaskRepository interface {
// Update operations
UpdateTaskStatus(ctx context.Context, taskID uuid.UUID, status string) error
+ // CancelOverduePendingTasks marks pending tasks of the given type whose
+ // scheduled_at is more than `overdue` in the past as cancelled. A pending
+ // task that far past its slot means the Cloud Tasks callback was lost
+ // (queue wipe, emulator restart, dropped retry); cancelling unblocks the
+ // reconcilers' "no pending task" checks so the chain re-seeds. A late
+ // callback for a cancelled row is a no-op (handlers require 'pending').
+ CancelOverduePendingTasks(ctx context.Context, taskType string, overdue time.Duration) (int64, error)
UpdateTaskScheduledAt(ctx context.Context, taskID uuid.UUID, scheduledAt time.Time, cloudTaskName string) error
RecordTaskFailure(ctx context.Context, taskID uuid.UUID, title, message string) error
@@ -544,7 +551,7 @@ func (r *taskRepository) GetScheduledTasksToday(ctx context.Context, accountID u
func (r *taskRepository) UpdateTaskStatus(ctx context.Context, taskID uuid.UUID, status string) error {
query := `
UPDATE tasks
- SET status = $1,
+ SET status = $1::task_status,
updated_at = NOW(),
completed_at = CASE WHEN $1 = 'completed' THEN NOW() ELSE completed_at END
WHERE id = $2
@@ -554,6 +561,23 @@ func (r *taskRepository) UpdateTaskStatus(ctx context.Context, taskID uuid.UUID,
return err
}
+// CancelOverduePendingTasks cancels pending tasks stranded past their slot;
+// see the interface comment for why.
+func (r *taskRepository) CancelOverduePendingTasks(ctx context.Context, taskType string, overdue time.Duration) (int64, error) {
+ query := `
+ UPDATE tasks
+ SET status = 'cancelled', updated_at = NOW()
+ WHERE task_type = $1
+ AND status = 'pending'
+ AND scheduled_at < NOW() - $2::interval
+ `
+ tag, err := r.db.Exec(ctx, query, taskType, overdue.String())
+ if err != nil {
+ return 0, err
+ }
+ return tag.RowsAffected(), nil
+}
+
// UpdateTaskScheduledAt updates the scheduled time and cloud task name
func (r *taskRepository) UpdateTaskScheduledAt(ctx context.Context, taskID uuid.UUID, scheduledAt time.Time, cloudTaskName string) error {
query := `
@@ -718,7 +742,7 @@ func (r *taskRepository) UpdateTaskStatusWithLock(ctx context.Context, taskID uu
// Update status
query := `
UPDATE tasks
- SET status = $1,
+ SET status = $1::task_status,
updated_at = NOW(),
completed_at = CASE WHEN $1 = 'completed' THEN NOW() ELSE completed_at END
WHERE id = $2
diff --git a/internal/repository/pg_worker.go b/internal/repository/pg_worker.go
index 8fabe5b3..ff4c6372 100644
--- a/internal/repository/pg_worker.go
+++ b/internal/repository/pg_worker.go
@@ -43,7 +43,7 @@ type WorkerRepository interface {
CreateDedicatedAssignment(ctx context.Context, assignment *models.DedicatedWorkerAssignment) error
CreateDedicatedAssignmentIfNotExists(ctx context.Context, assignment *models.DedicatedWorkerAssignment) (bool, error)
GetActiveDedicatedAssignment(ctx context.Context, userID uuid.UUID) (*models.DedicatedWorkerAssignment, error)
- GetDedicatedWorkerByUserID(ctx context.Context, userID uuid.UUID) (*models.Worker, error)
+ GetDedicatedWorkerByOrgID(ctx context.Context, orgID uuid.UUID) (*models.Worker, error)
ReleaseDedicatedAssignment(ctx context.Context, userID uuid.UUID) error
// Email account worker queries
@@ -282,14 +282,14 @@ func (r *workerRepository) SetWorkerType(ctx context.Context, workerID uuid.UUID
// CreateDedicatedAssignment creates a new dedicated worker assignment
func (r *workerRepository) CreateDedicatedAssignment(ctx context.Context, assignment *models.DedicatedWorkerAssignment) error {
query := `
- INSERT INTO dedicated_worker_assignments (id, worker_id, user_id, subscription_id, assigned_at)
+ INSERT INTO dedicated_worker_assignments (id, worker_id, organization_id, subscription_id, assigned_at)
VALUES ($1, $2, $3, $4, $5)
`
_, err := r.db.Exec(ctx, query,
assignment.ID,
assignment.WorkerID,
- assignment.UserID,
+ assignment.OrganizationID,
assignment.SubscriptionID,
assignment.AssignedAt,
)
@@ -297,21 +297,21 @@ func (r *workerRepository) CreateDedicatedAssignment(ctx context.Context, assign
}
// CreateDedicatedAssignmentIfNotExists atomically creates a dedicated worker assignment
-// only if no active (released_at IS NULL) assignment exists for the user.
+// only if no active (released_at IS NULL) assignment exists for the organization.
// Returns (true, nil) if created, (false, nil) if already exists.
func (r *workerRepository) CreateDedicatedAssignmentIfNotExists(ctx context.Context, assignment *models.DedicatedWorkerAssignment) (bool, error) {
query := `
- INSERT INTO dedicated_worker_assignments (id, worker_id, user_id, subscription_id, assigned_at)
+ INSERT INTO dedicated_worker_assignments (id, worker_id, organization_id, subscription_id, assigned_at)
SELECT $1, $2, $3, $4, $5
WHERE NOT EXISTS (
SELECT 1 FROM dedicated_worker_assignments
- WHERE user_id = $3 AND released_at IS NULL
+ WHERE organization_id = $3 AND released_at IS NULL
)
`
result, err := r.db.Exec(ctx, query,
assignment.ID,
assignment.WorkerID,
- assignment.UserID,
+ assignment.OrganizationID,
assignment.SubscriptionID,
assignment.AssignedAt,
)
@@ -321,17 +321,17 @@ func (r *workerRepository) CreateDedicatedAssignmentIfNotExists(ctx context.Cont
return result.RowsAffected() > 0, nil
}
-// GetActiveDedicatedAssignment retrieves the active dedicated assignment for a user
+// GetActiveDedicatedAssignment retrieves the active dedicated assignment for an organization
func (r *workerRepository) GetActiveDedicatedAssignment(ctx context.Context, userID uuid.UUID) (*models.DedicatedWorkerAssignment, error) {
query := `
- SELECT id, worker_id, user_id, subscription_id, assigned_at, released_at
+ SELECT id, worker_id, organization_id, subscription_id, assigned_at, released_at
FROM dedicated_worker_assignments
- WHERE user_id = $1 AND released_at IS NULL
+ WHERE organization_id = $1 AND released_at IS NULL
`
var a models.DedicatedWorkerAssignment
err := r.db.QueryRow(ctx, query, userID).Scan(
- &a.ID, &a.WorkerID, &a.UserID, &a.SubscriptionID, &a.AssignedAt, &a.ReleasedAt,
+ &a.ID, &a.WorkerID, &a.OrganizationID, &a.SubscriptionID, &a.AssignedAt, &a.ReleasedAt,
)
if err == pgx.ErrNoRows {
return nil, nil
@@ -342,17 +342,17 @@ func (r *workerRepository) GetActiveDedicatedAssignment(ctx context.Context, use
return &a, nil
}
-// GetDedicatedWorkerByUserID retrieves the dedicated worker assigned to a user
-func (r *workerRepository) GetDedicatedWorkerByUserID(ctx context.Context, userID uuid.UUID) (*models.Worker, error) {
+// GetDedicatedWorkerByOrgID retrieves the dedicated worker assigned to an organization
+func (r *workerRepository) GetDedicatedWorkerByOrgID(ctx context.Context, orgID uuid.UUID) (*models.Worker, error) {
query := `
SELECT w.id, w.ip_addr, w.active, w.free_tier, w.worker_type, w.account_count, w.created_at, w.updated_at
FROM workers w
JOIN dedicated_worker_assignments dwa ON w.id = dwa.worker_id
- WHERE dwa.user_id = $1 AND dwa.released_at IS NULL
+ WHERE dwa.organization_id = $1 AND dwa.released_at IS NULL
`
var w models.Worker
- err := r.db.QueryRow(ctx, query, userID).Scan(
+ err := r.db.QueryRow(ctx, query, orgID).Scan(
&w.ID, &w.IPAddr, &w.Active, &w.FreeTier, &w.WorkerType, &w.AccountCount,
&w.CreatedAt, &w.UpdatedAt,
)
@@ -370,7 +370,7 @@ func (r *workerRepository) ReleaseDedicatedAssignment(ctx context.Context, userI
query := `
UPDATE dedicated_worker_assignments
SET released_at = $1
- WHERE user_id = $2 AND released_at IS NULL
+ WHERE organization_id = $2 AND released_at IS NULL
`
_, err := r.db.Exec(ctx, query, time.Now(), userID)
diff --git a/internal/sandbox/config.go b/internal/sandbox/config.go
new file mode 100644
index 00000000..aeed318a
--- /dev/null
+++ b/internal/sandbox/config.go
@@ -0,0 +1,78 @@
+// Package sandbox seeds and animates a fully working local demo environment:
+// a paid showcase org whose mailboxes really send (SMTP -> mailpit) and really
+// sync (IMAP <- dovecot), plus a simulator that plays "the internet" - it
+// routes captured mail into recipient inboxes, opens tracking pixels, clicks
+// tracked links, and writes replies as the seeded contacts. Every platform
+// code path involved (scheduler, worker, consumer, tracking, realtime) is the
+// production one; only the humans are simulated.
+//
+// See docs/content/docs/development/sandbox.mdx for the full walkthrough.
+package sandbox
+
+import (
+ "os"
+ "strconv"
+)
+
+// Config carries the endpoints the seeder and simulator talk to. Defaults
+// match the `make infra` host-published ports.
+type Config struct {
+ // DatabaseURL is the dev Postgres DSN (seeding + entity lookups).
+ DatabaseURL string
+
+ // MailpitURL is the mailpit HTTP API base (message capture).
+ MailpitURL string
+
+ // TrackingURL is the tracking service base for pixel/click hits.
+ TrackingURL string
+
+ // IMAPAddr is the dovecot IMAPS address the SIMULATOR appends mail to.
+ IMAPAddr string
+ // IMAPPassword is dovecot's static password (any username works).
+ IMAPPassword string
+
+ // SMTPHost/SMTPPort are seeded into sandbox mailboxes as their outbound
+ // server; the WORKER dials these, so they are host-relative (mailpit).
+ SMTPHost string
+ SMTPPort int
+ // IMAPHost/IMAPPort are seeded as the mailboxes' inbound server; the
+ // WORKER dials these too (dovecot IMAPS).
+ IMAPHost string
+ IMAPPort int
+
+ // CredentialsKey is the CREDENTIALS_ENCRYPTION_KEY hex used to seal the
+ // seeded SMTP/IMAP credentials; must match the backend's key.
+ CredentialsKey string
+}
+
+// FromEnv builds a Config from the environment with `make infra` defaults.
+func FromEnv() Config {
+ return Config{
+ DatabaseURL: getenv("PRIMARY_DB", "postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable"),
+ MailpitURL: getenv("MAILPIT_URL", "http://localhost:18025"),
+ TrackingURL: getenv("TRACKING_URL", "http://localhost:3000"),
+ IMAPAddr: getenv("DOVECOT_IMAP_ADDR", "localhost:10993"),
+ IMAPPassword: getenv("DOVECOT_PASSWORD", "sandbox"),
+ SMTPHost: getenv("SANDBOX_SMTP_HOST", "localhost"),
+ SMTPPort: getenvInt("SANDBOX_SMTP_PORT", 11025),
+ IMAPHost: getenv("SANDBOX_IMAP_HOST", "localhost"),
+ IMAPPort: getenvInt("SANDBOX_IMAP_PORT", 10993),
+ CredentialsKey: os.Getenv("CREDENTIALS_ENCRYPTION_KEY"),
+ }
+}
+
+func getenv(key, fallback string) string {
+ if v := os.Getenv(key); v != "" {
+ return v
+ }
+ return fallback
+}
+
+func getenvInt(key string, fallback int) int {
+ if v := os.Getenv(key); v != "" {
+ if n, err := strconv.Atoi(v); err == nil {
+ return n
+ }
+ }
+ return fallback
+}
diff --git a/internal/sandbox/deliver.go b/internal/sandbox/deliver.go
new file mode 100644
index 00000000..fed7a83c
--- /dev/null
+++ b/internal/sandbox/deliver.go
@@ -0,0 +1,86 @@
+package sandbox
+
+import (
+ "crypto/tls"
+ "fmt"
+ "strings"
+ "time"
+
+ "github.com/emersion/go-imap/v2/imapclient"
+ "github.com/google/uuid"
+)
+
+// deliverToInbox appends a raw RFC822 message into a dovecot user's INBOX.
+// This is the sandbox's "final delivery" hop: mail captured by mailpit is
+// placed where the worker's real IMAP sync will find it. One short-lived
+// connection per delivery keeps the client trivially correct.
+func deliverToInbox(imapAddr, user, password string, raw []byte) error {
+ c, err := imapclient.DialTLS(imapAddr, &imapclient.Options{
+ // Dovecot's cert is self-signed; this client only ever talks to the
+ // local sandbox container.
+ TLSConfig: &tls.Config{InsecureSkipVerify: true},
+ })
+ if err != nil {
+ return fmt.Errorf("imap dial %s: %w", imapAddr, err)
+ }
+ defer c.Close()
+
+ if err := c.Login(user, password).Wait(); err != nil {
+ return fmt.Errorf("imap login %s: %w", user, err)
+ }
+
+ cmd := c.Append("INBOX", int64(len(raw)), nil)
+ if _, err := cmd.Write(raw); err != nil {
+ return fmt.Errorf("imap append write: %w", err)
+ }
+ if err := cmd.Close(); err != nil {
+ return fmt.Errorf("imap append close: %w", err)
+ }
+ if _, err := cmd.Wait(); err != nil {
+ return fmt.Errorf("imap append: %w", err)
+ }
+ if err := c.Logout().Wait(); err != nil {
+ return fmt.Errorf("imap logout: %w", err)
+ }
+ return nil
+}
+
+// composeReply builds the RFC822 reply a contact sends back to a sandbox
+// mailbox. In-Reply-To/References carry the original Message-ID so the
+// consumer's reply attribution (tasks.message_id lookup) resolves, and
+// automated replies carry Auto-Submitted so the reply classifier gates them
+// out of human-reply stats.
+func composeReply(fromName, fromAddr, toName, toAddr, subject, origMessageID, body string, automated bool) []byte {
+ if !strings.HasPrefix(strings.ToLower(subject), "re:") {
+ subject = "Re: " + subject
+ }
+ mid := fmt.Sprintf("<%s@%s>", uuid.New().String(), domainOf(fromAddr))
+ orig := "<" + strings.Trim(origMessageID, "<>") + ">"
+
+ var b strings.Builder
+ fmt.Fprintf(&b, "From: %q <%s>\r\n", fromName, fromAddr)
+ fmt.Fprintf(&b, "To: %q <%s>\r\n", toName, toAddr)
+ fmt.Fprintf(&b, "Subject: %s\r\n", subject)
+ fmt.Fprintf(&b, "Message-ID: %s\r\n", mid)
+ fmt.Fprintf(&b, "In-Reply-To: %s\r\n", orig)
+ fmt.Fprintf(&b, "References: %s\r\n", orig)
+ fmt.Fprintf(&b, "Date: %s\r\n", time.Now().Format(time.RFC1123Z))
+ if automated {
+ b.WriteString("Auto-Submitted: auto-replied\r\n")
+ b.WriteString("X-Autoreply: yes\r\n")
+ b.WriteString("Precedence: auto_reply\r\n")
+ }
+ b.WriteString("MIME-Version: 1.0\r\n")
+ b.WriteString("Content-Type: text/plain; charset=UTF-8\r\n")
+ b.WriteString("\r\n")
+ b.WriteString(strings.ReplaceAll(body, "\n", "\r\n"))
+ b.WriteString("\r\n")
+ return []byte(b.String())
+}
+
+func domainOf(addr string) string {
+ if at := strings.LastIndex(addr, "@"); at >= 0 {
+ return addr[at+1:]
+ }
+ return "sandbox.test"
+}
diff --git a/internal/sandbox/mailpit.go b/internal/sandbox/mailpit.go
new file mode 100644
index 00000000..371647c1
--- /dev/null
+++ b/internal/sandbox/mailpit.go
@@ -0,0 +1,127 @@
+package sandbox
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "time"
+)
+
+// mailpitClient is a minimal client for the mailpit HTTP API; the simulator
+// uses the Read flag as its processing cursor.
+type mailpitClient struct {
+ base string
+ http *http.Client
+}
+
+func newMailpitClient(base string) *mailpitClient {
+ return &mailpitClient{base: base, http: &http.Client{Timeout: 15 * time.Second}}
+}
+
+type mailpitAddress struct {
+ Name string `json:"Name"`
+ Address string `json:"Address"`
+}
+
+type mailpitSummary struct {
+ ID string `json:"ID"`
+ Read bool `json:"Read"`
+ From mailpitAddress `json:"From"`
+ To []mailpitAddress `json:"To"`
+ Subject string `json:"Subject"`
+}
+
+type mailpitListResponse struct {
+ Messages []mailpitSummary `json:"messages"`
+}
+
+type mailpitMessage struct {
+ ID string `json:"ID"`
+ MessageID string `json:"MessageID"`
+ From mailpitAddress `json:"From"`
+ To []mailpitAddress `json:"To"`
+ Subject string `json:"Subject"`
+ HTML string `json:"HTML"`
+ Text string `json:"Text"`
+}
+
+// listUnread returns the newest messages that have not been marked read.
+func (c *mailpitClient) listUnread(ctx context.Context, limit int) ([]mailpitSummary, error) {
+ var out mailpitListResponse
+ if err := c.get(ctx, fmt.Sprintf("/api/v1/messages?limit=%d", limit), &out); err != nil {
+ return nil, err
+ }
+ unread := out.Messages[:0]
+ for _, m := range out.Messages {
+ if !m.Read {
+ unread = append(unread, m)
+ }
+ }
+ return unread, nil
+}
+
+func (c *mailpitClient) message(ctx context.Context, id string) (*mailpitMessage, error) {
+ var out mailpitMessage
+ if err := c.get(ctx, "/api/v1/message/"+id, &out); err != nil {
+ return nil, err
+ }
+ return &out, nil
+}
+
+// raw returns the full RFC822 source of a captured message.
+func (c *mailpitClient) raw(ctx context.Context, id string) ([]byte, error) {
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+"/api/v1/message/"+id+"/raw", nil)
+ if err != nil {
+ return nil, err
+ }
+ resp, err := c.http.Do(req)
+ if err != nil {
+ return nil, err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return nil, fmt.Errorf("mailpit raw %s: status %d", id, resp.StatusCode)
+ }
+ return io.ReadAll(resp.Body)
+}
+
+// markRead flips the Read flag; this is the simulator's "processed" cursor.
+func (c *mailpitClient) markRead(ctx context.Context, ids []string) error {
+ body, err := json.Marshal(map[string]any{"IDs": ids, "Read": true})
+ if err != nil {
+ return err
+ }
+ req, err := http.NewRequestWithContext(ctx, http.MethodPut, c.base+"/api/v1/messages", bytes.NewReader(body))
+ if err != nil {
+ return err
+ }
+ req.Header.Set("Content-Type", "application/json")
+ resp, err := c.http.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("mailpit mark read: status %d", resp.StatusCode)
+ }
+ return nil
+}
+
+func (c *mailpitClient) get(ctx context.Context, path string, out any) error {
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+path, nil)
+ if err != nil {
+ return err
+ }
+ resp, err := c.http.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("mailpit GET %s: status %d", path, resp.StatusCode)
+ }
+ return json.NewDecoder(resp.Body).Decode(out)
+}
diff --git a/internal/sandbox/personas.go b/internal/sandbox/personas.go
new file mode 100644
index 00000000..fd44cdf3
--- /dev/null
+++ b/internal/sandbox/personas.go
@@ -0,0 +1,75 @@
+package sandbox
+
+import (
+ "fmt"
+ "hash/fnv"
+)
+
+// persona is a contact's deterministic behavior profile, derived from the
+// email address so it is stable across simulator restarts with no DB state.
+type persona struct {
+ Opens bool
+ Clicks bool
+ Replies bool
+ Flavor replyFlavor
+}
+
+type replyFlavor int
+
+const (
+ replyPositive replyFlavor = iota
+ replyQuestion
+ replyNegative
+ replyOutOfOffice
+)
+
+// personaFor buckets a contact by hash: most open, some click, some reply,
+// and reply tone varies so the reply classifier has real work to do.
+func personaFor(email string) persona {
+ h := fnv.New32a()
+ h.Write([]byte(email))
+ n := h.Sum32()
+
+ p := persona{
+ Opens: n%100 < 85,
+ Clicks: n%7 < 3, // ~43% of openers
+ Replies: n%11 < 4, // ~36% of openers
+ }
+ switch n % 10 {
+ case 0, 1, 2, 3:
+ p.Flavor = replyPositive
+ case 4, 5:
+ p.Flavor = replyQuestion
+ case 6, 7:
+ p.Flavor = replyNegative
+ default:
+ p.Flavor = replyOutOfOffice
+ }
+ return p
+}
+
+// replyBody returns the reply text plus whether the reply is an automated one
+// (out-of-office), which gets auto-reply headers so the classifier can gate it
+// out of human-reply stats.
+func replyBody(f replyFlavor, firstName string) (body string, automated bool) {
+ switch f {
+ case replyPositive:
+ return fmt.Sprintf("Hi,\n\nThis looks interesting. Can you send over pricing and a couple of customer references?\n\nThanks,\n%s", firstName), false
+ case replyQuestion:
+ return fmt.Sprintf("Hi,\n\nBefore we go further: does this integrate with our existing CRM, and where is the data hosted?\n\n%s", firstName), false
+ case replyNegative:
+ return fmt.Sprintf("Hi,\n\nNot a fit for us right now. Please remove me from this list.\n\n%s", firstName), false
+ default:
+ return "Hello,\n\nI am currently out of the office with limited access to email and will respond on my return.\n\nThis is an automated response.", true
+ }
+}
+
+// userAgents rotate across opens/clicks so tracking data looks organic. None
+// of these match the tracking service's prefetch/scanner filter list.
+var userAgents = []string{
+ "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
+ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
+ "Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Mobile/15E148 Safari/604.1",
+ "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.0",
+ "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Safari/605.1.15",
+}
diff --git a/internal/sandbox/seed.go b/internal/sandbox/seed.go
new file mode 100644
index 00000000..40903a4e
--- /dev/null
+++ b/internal/sandbox/seed.go
@@ -0,0 +1,526 @@
+package sandbox
+
+import (
+ "context"
+ "fmt"
+ "regexp"
+ "strings"
+
+ "github.com/google/uuid"
+ "github.com/jackc/pgx/v5/pgxpool"
+ "github.com/warmbly/warmbly/internal/pkg/argon2"
+ "github.com/warmbly/warmbly/internal/pkg/encrypt"
+ "github.com/warmbly/warmbly/internal/seed"
+)
+
+// Stable UUIDs (middle group "aaaa" marks sandbox rows; disjoint from the
+// cmd/seed and internal/seed namespaces).
+var (
+ sandboxUser = uuid.MustParse("11111111-aaaa-0000-0000-000000000001")
+ sandboxOrg = uuid.MustParse("22222222-aaaa-0000-0000-000000000001")
+ sandboxSub = uuid.MustParse("88888888-aaaa-0000-0000-000000000001")
+
+ // The premium shared worker (`make worker-premium` natively, or the
+ // docker worker-premium-1). Paid orgs place strictly onto premium
+ // workers, so the sandbox mailboxes must live here.
+ sandboxWorker = uuid.MustParse("10c8f5e4-1c39-5b2a-9c8b-3d2f0a8b1a02")
+
+ campaignLaunch = uuid.MustParse("44444444-aaaa-0000-0000-000000000001")
+ campaignAgency = uuid.MustParse("44444444-aaaa-0000-0000-000000000002")
+ campaignDormant = uuid.MustParse("44444444-aaaa-0000-0000-000000000003")
+)
+
+// SandboxLoginEmail / SandboxLoginPassword are the dashboard credentials the
+// seeder prints; keep in sync with docs/content/docs/development/sandbox.mdx.
+const (
+ SandboxLoginEmail = "sandbox@warmbly.test"
+ SandboxLoginPassword = "password123"
+)
+
+type mailboxSeed struct {
+ id uuid.UUID
+ email string
+ name string
+}
+
+// sandboxMailboxes are the org's senders, all hosted on the local dovecot.
+var sandboxMailboxes = []mailboxSeed{
+ {uuid.MustParse("33333333-aaaa-0000-0000-000000000001"), "sarah.lin@sunrise.test", "Sarah Lin"},
+ {uuid.MustParse("33333333-aaaa-0000-0000-000000000002"), "marcus.reid@sunrise.test", "Marcus Reid"},
+ {uuid.MustParse("33333333-aaaa-0000-0000-000000000003"), "priya.nair@sunrise.test", "Priya Nair"},
+ {uuid.MustParse("33333333-aaaa-0000-0000-000000000004"), "tom.abel@sunrise.test", "Tom Abel"},
+ {uuid.MustParse("33333333-aaaa-0000-0000-000000000005"), "elena.voss@sunrise.test", "Elena Voss"},
+ {uuid.MustParse("33333333-aaaa-0000-0000-000000000006"), "dan.okafor@sunrise.test", "Dan Okafor"},
+}
+
+type contactSeed struct {
+ first, last, email, company string
+ subscribed bool
+}
+
+// Launch-campaign prospects. Two are unsubscribed so suppression shows up.
+var launchContacts = []contactSeed{
+ {"Aiden", "Park", "aiden.park@northwind.test", "Northwind", true},
+ {"Beth", "Chen", "beth.chen@initech.test", "Initech", true},
+ {"Carlos", "Diaz", "carlos.diaz@piedpiper.test", "Pied Piper", true},
+ {"Diana", "Fox", "diana.fox@globex.test", "Globex", true},
+ {"Eli", "Grant", "eli.grant@hooli.test", "Hooli", true},
+ {"Fiona", "Hale", "fiona.hale@umbrella.test", "Umbrella", false},
+ {"Greg", "Iver", "greg.iver@stark.test", "Stark Industries", true},
+ {"Hana", "Jules", "hana.jules@wayne.test", "Wayne Enterprises", true},
+ {"Ivan", "Kova", "ivan.kova@tyrell.test", "Tyrell", true},
+ {"Jade", "Lund", "jade.lund@wonka.test", "Wonka", true},
+ {"Kofi", "Mensah", "kofi.mensah@acme-corp.test", "Acme Corp", true},
+ {"Lena", "Novak", "lena.novak@cyberdyne.test", "Cyberdyne", true},
+ {"Mia", "Ono", "mia.ono@soylent.test", "Soylent", true},
+ {"Nils", "Pett", "nils.pett@aperture.test", "Aperture", false},
+ {"Olga", "Quist", "olga.quist@blackmesa.test", "Black Mesa", true},
+ {"Pablo", "Rey", "pablo.rey@monsters.test", "Monsters Inc", true},
+ {"Quinn", "Soto", "quinn.soto@dunder.test", "Dunder Mifflin", true},
+ {"Rita", "Tam", "rita.tam@vandelay.test", "Vandelay", true},
+ {"Sam", "Ueda", "sam.ueda@prestige.test", "Prestige Worldwide", true},
+ {"Tara", "Vale", "tara.vale@oceanic.test", "Oceanic", true},
+ {"Umar", "Wolf", "umar.wolf@massive.test", "Massive Dynamic", true},
+ {"Vera", "Xu", "vera.xu@virtucon.test", "Virtucon", true},
+ {"Wes", "York", "wes.york@octan.test", "Octan", true},
+ {"Xena", "Zair", "xena.zair@gringotts.test", "Gringotts", true},
+}
+
+// Agency-campaign prospects.
+var agencyContacts = []contactSeed{
+ {"Amara", "Bell", "amara.bell@brightloop.test", "Brightloop Agency", true},
+ {"Boris", "Chan", "boris.chan@funnelworks.test", "Funnelworks", true},
+ {"Cleo", "Danes", "cleo.danes@leadcraft.test", "Leadcraft", true},
+ {"Derek", "Enns", "derek.enns@growthlab.test", "Growthlab", true},
+ {"Esme", "Ford", "esme.ford@pipelinehq.test", "Pipeline HQ", true},
+ {"Felix", "Gaunt", "felix.gaunt@outbounders.test", "Outbounders", true},
+ {"Gita", "Hart", "gita.hart@replyrate.test", "Replyrate", true},
+ {"Hugo", "Ines", "hugo.ines@coldsmiths.test", "Coldsmiths", true},
+ {"Iris", "Joon", "iris.joon@meetingmakers.test", "Meeting Makers", true},
+ {"Jonas", "Kemp", "jonas.kemp@quotaquest.test", "Quotaquest", true},
+ {"Kira", "Lowe", "kira.lowe@demodesk.test", "Demodesk Partners", true},
+ {"Liam", "Moss", "liam.moss@sequoialeads.test", "Sequoia Leads", true},
+}
+
+// Seed provisions the sandbox: the full internal/seed fixture (plans, demo
+// orgs, workers), then the sandbox org with live mailboxes, active campaigns,
+// contacts, warmup membership, and a paid subscription. Finally it repairs
+// EVERY smtp_imap account's credentials to point at mailpit/dovecot so the
+// whole warmup pool can actually send and sync. Idempotent throughout.
+func Seed(ctx context.Context, pool *pgxpool.Pool, cfg Config) error {
+ if cfg.CredentialsKey == "" {
+ return fmt.Errorf("CREDENTIALS_ENCRYPTION_KEY is required to seed working mailboxes")
+ }
+ enc, err := encrypt.NewEncrypterFromHex(cfg.CredentialsKey)
+ if err != nil {
+ return fmt.Errorf("CREDENTIALS_ENCRYPTION_KEY: %w", err)
+ }
+
+ // Base fixture: plans + durations + workers + the two demo orgs. Running
+ // it here makes `make sandbox` self-contained on a fresh database.
+ if _, err := seed.Run(ctx, pool); err != nil {
+ return fmt.Errorf("base seed: %w", err)
+ }
+
+ if err := seedIdentity(ctx, pool); err != nil {
+ return err
+ }
+ if err := seedMailboxes(ctx, pool); err != nil {
+ return err
+ }
+ if err := seedSubscription(ctx, pool); err != nil {
+ return err
+ }
+ if err := seedCampaigns(ctx, pool); err != nil {
+ return err
+ }
+ if err := repairSMTPIMAPCredentials(ctx, pool, cfg, enc); err != nil {
+ return err
+ }
+ if err := repairContactVerification(ctx, pool); err != nil {
+ return err
+ }
+ if err := deactivateIdleFixtureWorkers(ctx, pool); err != nil {
+ return err
+ }
+
+ fmt.Println("sandbox seeded:")
+ fmt.Printf(" dashboard %s / %s (org: Sunrise Labs)\n", SandboxLoginEmail, SandboxLoginPassword)
+ fmt.Printf(" mailboxes %d senders on @sunrise.test (SMTP -> mailpit, IMAP -> dovecot)\n", len(sandboxMailboxes))
+ fmt.Println(" campaigns Sunrise Q3 launch + Agency partnerships (active), Dormant reactivation (draft)")
+ fmt.Println(" warmup enabled on all senders, premium pool")
+ return nil
+}
+
+func seedIdentity(ctx context.Context, pool *pgxpool.Pool) error {
+ var exists bool
+ if err := pool.QueryRow(ctx, "SELECT EXISTS(SELECT 1 FROM users WHERE id=$1)", sandboxUser).Scan(&exists); err != nil {
+ return err
+ }
+ if !exists {
+ hash, err := argon2.Hash(SandboxLoginPassword)
+ if err != nil {
+ return err
+ }
+ if _, err := pool.Exec(ctx, `
+ INSERT INTO users (id, first_name, last_name, email, password_hash)
+ VALUES ($1, 'Sunny', 'Sandbox', $2, $3)
+ ON CONFLICT (id) DO NOTHING`,
+ sandboxUser, SandboxLoginEmail, hash); err != nil {
+ return err
+ }
+ }
+
+ if _, err := pool.Exec(ctx, `
+ INSERT INTO organizations (id, name, slug, owner_user_id)
+ VALUES ($1, 'Sunrise Labs', 'sunrise-sandbox', $2)
+ ON CONFLICT (id) DO NOTHING`,
+ sandboxOrg, sandboxUser); err != nil {
+ return err
+ }
+ _, err := pool.Exec(ctx, `
+ INSERT INTO organization_members (organization_id, user_id, role, accepted_at)
+ VALUES ($1, $2, 'owner', NOW())
+ ON CONFLICT DO NOTHING`,
+ sandboxOrg, sandboxUser)
+ return err
+}
+
+func seedMailboxes(ctx context.Context, pool *pgxpool.Pool) error {
+ for _, m := range sandboxMailboxes {
+ // Warmup started 10 days ago so ramp progression is mid-flight; the
+ // send window is wide open and pacing is demo-friendly (90s min gap).
+ if _, err := pool.Exec(ctx, `
+ INSERT INTO email_accounts (
+ id, user_id, organization_id, worker_id,
+ email, name, signature_plain, signature_html,
+ provider, status,
+ campaign_limit, min_wait_time, timezone,
+ warmup, warmup_tag, warmup_pool_type,
+ warmup_start_time, warmup_end_time
+ ) VALUES (
+ $1, $2, $3, $4,
+ $5, $6, '', '',
+ 'smtp_imap', 'active',
+ 100, 90, 'UTC',
+ NOW() - INTERVAL '10 days', 'sandbox', 'premium',
+ '00:00', '23:59'
+ )
+ ON CONFLICT (id) DO UPDATE SET
+ worker_id = EXCLUDED.worker_id,
+ status = 'active',
+ campaign_limit = EXCLUDED.campaign_limit,
+ min_wait_time = EXCLUDED.min_wait_time,
+ warmup = COALESCE(email_accounts.warmup, EXCLUDED.warmup),
+ warmup_paused_at = NULL,
+ updated_at = NOW()`,
+ m.id, sandboxUser, sandboxOrg, sandboxWorker, m.email, m.name); err != nil {
+ return fmt.Errorf("mailbox %s: %w", m.email, err)
+ }
+ if _, err := pool.Exec(ctx, `
+ INSERT INTO warmup_pool_participants (pool_id, email_account_id)
+ SELECT id, $1 FROM warmup_pools WHERE pool_type = 'premium'::warmup_pool_type
+ ON CONFLICT DO NOTHING`,
+ m.id); err != nil {
+ return fmt.Errorf("pool join %s: %w", m.email, err)
+ }
+ }
+ return nil
+}
+
+func seedSubscription(ctx context.Context, pool *pgxpool.Pool) error {
+ _, err := pool.Exec(ctx, `
+ INSERT INTO subscriptions (
+ id, user_id, organization_id, plan_id,
+ stripe_customer_id, stripe_subscription_id, stripe_price_id,
+ status, current_period_start, current_period_end,
+ is_enterprise, created_at, updated_at
+ ) VALUES (
+ $1, $2, $3, $4,
+ 'cus_sandbox', 'sub_sandbox_starter', 'price_sandbox_starter',
+ 'active', NOW(), NOW() + INTERVAL '30 days',
+ FALSE, NOW(), NOW()
+ )
+ ON CONFLICT (organization_id) DO UPDATE SET
+ plan_id = EXCLUDED.plan_id,
+ status = 'active',
+ stripe_subscription_id = EXCLUDED.stripe_subscription_id,
+ current_period_start = EXCLUDED.current_period_start,
+ current_period_end = EXCLUDED.current_period_end,
+ updated_at = NOW()`,
+ sandboxSub, sandboxUser, sandboxOrg, seed.PlanStarterID)
+ return err
+}
+
+type campaignSeed struct {
+ id uuid.UUID
+ name string
+ status string
+ steps []stepSeed
+ contacts []contactSeed
+ // contactBase is the deterministic UUID prefix for this campaign's contacts.
+ contactBase string
+}
+
+type stepSeed struct {
+ id uuid.UUID
+ name string
+ subject string
+ body string
+ waitAfter int
+}
+
+func seedCampaigns(ctx context.Context, pool *pgxpool.Pool) error {
+ campaigns := []campaignSeed{
+ {
+ id: campaignLaunch, name: "Sunrise Q3 launch outreach", status: "active",
+ contactBase: "66666666-aaaa-0000-0001",
+ contacts: launchContacts,
+ steps: []stepSeed{
+ {uuid.MustParse("55555555-aaaa-0000-0000-000000000011"), "Intro",
+ "Quick question about {{.Company}}",
+ "Hi {{.FirstName}},\n\nWe just launched a tool that cuts outbound setup from weeks to minutes, and {{.Company}} came up twice in customer calls last month.\n\nWorth a quick look? Here is a two minute overview: https://warmbly.com/overview\n\nBest,\nSunrise team", 0},
+ {uuid.MustParse("55555555-aaaa-0000-0000-000000000012"), "Follow-up",
+ "Re: Quick question about {{.Company}}",
+ "Hi {{.FirstName}},\n\nFloating this back up. Happy to share the deliverability numbers from the beta if useful: https://warmbly.com/benchmarks\n\nBest,\nSunrise team", 2},
+ {uuid.MustParse("55555555-aaaa-0000-0000-000000000013"), "Breakup",
+ "Closing the loop",
+ "Hi {{.FirstName}},\n\nSounds like the timing is off. I will stop here; if outbound comes back on the roadmap, you know where to find us.\n\nBest,\nSunrise team", 4},
+ },
+ },
+ {
+ id: campaignAgency, name: "Agency partnerships", status: "active",
+ contactBase: "66666666-aaaa-0000-0002",
+ contacts: agencyContacts,
+ steps: []stepSeed{
+ {uuid.MustParse("55555555-aaaa-0000-0000-000000000021"), "Partner intro",
+ "Partnering with {{.Company}}",
+ "Hi {{.FirstName}},\n\nWe work with agencies like {{.Company}} on white-label sending infrastructure. Margins are meaningfully better than reselling seats.\n\nOpen to a short call? Details: https://warmbly.com/partners\n\nBest,\nSunrise partnerships", 0},
+ {uuid.MustParse("55555555-aaaa-0000-0000-000000000022"), "Partner follow-up",
+ "Re: Partnering with {{.Company}}",
+ "Hi {{.FirstName}},\n\nOne more nudge; the partner program closes new slots at the end of the quarter.\n\nBest,\nSunrise partnerships", 3},
+ },
+ },
+ {
+ id: campaignDormant, name: "Dormant accounts reactivation", status: "draft",
+ contactBase: "66666666-aaaa-0000-0003",
+ steps: []stepSeed{
+ {uuid.MustParse("55555555-aaaa-0000-0000-000000000031"), "Win-back",
+ "We miss you at {{.Company}}",
+ "Hi {{.FirstName}},\n\nA lot has shipped since you last looked. Draft for review before this goes anywhere.\n\nBest,\nSunrise team", 0},
+ },
+ },
+ }
+
+ for _, c := range campaigns {
+ // Wide-open schedule (all days, 00:00-23:59) so the demo sends now,
+ // not at the next business-hours boundary. Tracking on for both.
+ if _, err := pool.Exec(ctx, `
+ INSERT INTO campaigns (
+ id, user_id, organization_id, name, description,
+ status, days, start_time, end_time, timezone,
+ open_tracking, link_tracking,
+ updated_at, created_at
+ ) VALUES (
+ $1, $2, $3, $4, 'Sandbox showcase campaign',
+ $5, 127, '00:00', '23:59', 'UTC',
+ TRUE, TRUE,
+ NOW(), NOW()
+ )
+ ON CONFLICT (id) DO UPDATE SET
+ status = EXCLUDED.status,
+ days = EXCLUDED.days,
+ start_time = EXCLUDED.start_time,
+ end_time = EXCLUDED.end_time,
+ updated_at = NOW()`,
+ c.id, sandboxUser, sandboxOrg, c.name, c.status); err != nil {
+ return fmt.Errorf("campaign %s: %w", c.name, err)
+ }
+
+ for i, s := range c.steps {
+ if _, err := pool.Exec(ctx, `
+ INSERT INTO sequences (
+ id, campaign_id, organization_id, name, subject,
+ body_plain, body_html, wait_after, position
+ ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+ ON CONFLICT (id) DO NOTHING`,
+ s.id, c.id, sandboxOrg, s.name, s.subject, s.body, plainToHTML(s.body), s.waitAfter, i); err != nil {
+ return fmt.Errorf("sequence %s: %w", s.name, err)
+ }
+ }
+
+ for i, ct := range c.contacts {
+ cid := uuid.MustParse(fmt.Sprintf("%s-%012d", c.contactBase, i+1))
+ // Pre-verified: .test domains have no MX, so the live verifier
+ // would mark them invalid and the pre-send gate would skip every
+ // send. A stamped verdict is final (the sweep only processes
+ // unchecked contacts).
+ if _, err := pool.Exec(ctx, `
+ INSERT INTO contacts (
+ id, user_id, organization_id,
+ first_name, last_name, email, company, phone,
+ custom_fields, subscribed,
+ verification_status, verification_reason, verification_checked_at
+ ) VALUES ($1, $2, $3, $4, $5, $6, $7, '', '{}', $8,
+ 'valid', 'sandbox fixture address', NOW())
+ ON CONFLICT (id) DO UPDATE SET
+ verification_status = 'valid',
+ verification_reason = 'sandbox fixture address',
+ verification_checked_at = NOW()`,
+ cid, sandboxUser, sandboxOrg, ct.first, ct.last, ct.email, ct.company, ct.subscribed); err != nil {
+ return fmt.Errorf("contact %s: %w", ct.email, err)
+ }
+ if _, err := pool.Exec(ctx, `
+ INSERT INTO campaign_leads (campaign_id, contact_id)
+ VALUES ($1, $2)
+ ON CONFLICT DO NOTHING`,
+ c.id, cid); err != nil {
+ return fmt.Errorf("campaign lead %s: %w", ct.email, err)
+ }
+ }
+ }
+ return nil
+}
+
+// repairSMTPIMAPCredentials points every smtp_imap account (sandbox AND the
+// cmd/seed / internal/seed fixtures, whose stored credentials are plaintext
+// placeholders) at the local mail stack, sealed with the credentials key so
+// the worker loader can decrypt and actually send/sync them.
+func repairSMTPIMAPCredentials(ctx context.Context, pool *pgxpool.Pool, cfg Config, enc *encrypt.Encrypter) error {
+ rows, err := pool.Query(ctx, `SELECT id, email FROM email_accounts WHERE provider = 'smtp_imap'`)
+ if err != nil {
+ return err
+ }
+ defer rows.Close()
+
+ type acct struct {
+ id uuid.UUID
+ email string
+ }
+ var accts []acct
+ for rows.Next() {
+ var a acct
+ if err := rows.Scan(&a.id, &a.email); err != nil {
+ return err
+ }
+ accts = append(accts, a)
+ }
+ if err := rows.Err(); err != nil {
+ return err
+ }
+
+ sealed := func(s string) (string, error) { return enc.Encrypt(s) }
+ for _, a := range accts {
+ smtpHost, err1 := sealed(cfg.SMTPHost)
+ smtpUser, err2 := sealed(a.email)
+ smtpPass, err3 := sealed(cfg.IMAPPassword)
+ imapHost, err4 := sealed(cfg.IMAPHost)
+ imapUser, err5 := sealed(a.email)
+ imapPass, err6 := sealed(cfg.IMAPPassword)
+ for _, e := range []error{err1, err2, err3, err4, err5, err6} {
+ if e != nil {
+ return e
+ }
+ }
+ if _, err := pool.Exec(ctx, `
+ INSERT INTO email_accounts_smtp_imap (
+ email_account_id,
+ smtp_host, smtp_port, smtp_user, smtp_password,
+ imap_host, imap_port, imap_user, imap_password
+ ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+ ON CONFLICT (email_account_id) DO UPDATE SET
+ smtp_host = EXCLUDED.smtp_host,
+ smtp_port = EXCLUDED.smtp_port,
+ smtp_user = EXCLUDED.smtp_user,
+ smtp_password = EXCLUDED.smtp_password,
+ imap_host = EXCLUDED.imap_host,
+ imap_port = EXCLUDED.imap_port,
+ imap_user = EXCLUDED.imap_user,
+ imap_password = EXCLUDED.imap_password`,
+ a.id, smtpHost, cfg.SMTPPort, smtpUser, smtpPass,
+ imapHost, cfg.IMAPPort, imapUser, imapPass); err != nil {
+ return fmt.Errorf("credentials for %s: %w", a.email, err)
+ }
+ }
+ fmt.Printf(" credentials repaired for %d smtp_imap accounts\n", len(accts))
+ return nil
+}
+
+// repairContactVerification marks every fixture contact (.test addresses,
+// including the cmd/seed and internal/seed ones) as verified. The live
+// verifier finds no MX for .test domains and flags them invalid, after which
+// the pre-send gate would skip every campaign send in the sandbox.
+func repairContactVerification(ctx context.Context, pool *pgxpool.Pool) error {
+ tag, err := pool.Exec(ctx, `
+ UPDATE contacts
+ SET verification_status = 'valid',
+ verification_reason = 'sandbox fixture address',
+ verification_checked_at = NOW()
+ WHERE email LIKE '%.test' AND verification_status <> 'valid'`)
+ if err != nil {
+ return err
+ }
+ if n := tag.RowsAffected(); n > 0 {
+ fmt.Printf(" verification repaired for %d fixture contacts\n", n)
+ }
+ return nil
+}
+
+// deactivateIdleFixtureWorkers deactivates every seeded worker except the two
+// the native stack actually runs (`make worker` / `make worker-premium`).
+// Fixture workers are seeded active but never heartbeat, so placement keeps
+// choosing them and the dead-worker sweep keeps draining them - an assignment
+// ping-pong that strands mailboxes mid-send. `make seed` re-activates them
+// for the docker `make sim` flow.
+func deactivateIdleFixtureWorkers(ctx context.Context, pool *pgxpool.Pool) error {
+ tag, err := pool.Exec(ctx, `
+ UPDATE workers SET active = FALSE, updated_at = NOW()
+ WHERE active AND id NOT IN ($1, $2)`,
+ uuid.MustParse("10c8f5e4-1c39-5b2a-9c8b-3d2f0a8b1a01"), sandboxWorker)
+ if err != nil {
+ return err
+ }
+ if n := tag.RowsAffected(); n > 0 {
+ fmt.Printf(" deactivated %d fixture workers not running in the native stack\n", n)
+ }
+ return nil
+}
+
+// plainToHTML renders the plaintext step body as minimal paragraph HTML,
+// linkifying bare URLs into anchors so tracking has both a pixel target and
+// hrefs to wrap into click tickets.
+func plainToHTML(body string) string {
+ html := ""
+ for _, para := range splitParagraphs(body) {
+ html += "" + linkifyURLs(para) + "
"
+ }
+ return html
+}
+
+var urlPattern = regexp.MustCompile(`https://[^\s<]+`)
+
+func linkifyURLs(s string) string {
+ return urlPattern.ReplaceAllString(s, `$0`)
+}
+
+func splitParagraphs(s string) []string {
+ var out []string
+ cur := ""
+ for _, line := range strings.Split(s, "\n") {
+ if line == "" {
+ if cur != "" {
+ out = append(out, cur)
+ cur = ""
+ }
+ continue
+ }
+ if cur != "" {
+ cur += "
"
+ }
+ cur += line
+ }
+ if cur != "" {
+ out = append(out, cur)
+ }
+ return out
+}
diff --git a/internal/sandbox/simulate.go b/internal/sandbox/simulate.go
new file mode 100644
index 00000000..af82c3a1
--- /dev/null
+++ b/internal/sandbox/simulate.go
@@ -0,0 +1,275 @@
+package sandbox
+
+import (
+ "context"
+ "fmt"
+ "math/rand"
+ "net/http"
+ "regexp"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/jackc/pgx/v5/pgxpool"
+)
+
+// Pixel and click-ticket paths as injected by internal/tasks/template.go. The
+// domain in the mail is the (unresolvable) TRACKING_DOMAIN, so the simulator
+// keeps only the path and hits the local tracking service directly.
+var (
+ pixelRe = regexp.MustCompile(`/t/o/([0-9a-fA-F-]{36})\.png`)
+ clickRe = regexp.MustCompile(`/c/([0-9a-fA-F-]{36})`)
+)
+
+type simulator struct {
+ cfg Config
+ pool *pgxpool.Pool
+ mp *mailpitClient
+ http *http.Client
+
+ mu sync.RWMutex
+ hosted map[string]hostedMailbox // lowercased address -> mailbox
+ contact map[string]contactInfo // lowercased address -> contact
+}
+
+type hostedMailbox struct {
+ Email string
+ Name string
+}
+
+type contactInfo struct {
+ Email string
+ First string
+ Last string
+}
+
+// Simulate runs the sandbox's "internet" until ctx is cancelled: it routes
+// captured mail into dovecot inboxes and plays the seeded contacts (opens,
+// clicks, replies). Safe to restart at any time; the mailpit Read flag is the
+// cursor, so nothing is processed twice.
+func Simulate(ctx context.Context, pool *pgxpool.Pool, cfg Config) error {
+ s := &simulator{
+ cfg: cfg,
+ pool: pool,
+ mp: newMailpitClient(cfg.MailpitURL),
+ http: &http.Client{
+ Timeout: 15 * time.Second,
+ // Click tickets 302 to their destination; the destination does not
+ // resolve locally, so record the click and stop at the redirect.
+ CheckRedirect: func(req *http.Request, via []*http.Request) error {
+ return http.ErrUseLastResponse
+ },
+ },
+ hosted: map[string]hostedMailbox{},
+ contact: map[string]contactInfo{},
+ }
+
+ if err := s.refreshDirectory(ctx); err != nil {
+ return fmt.Errorf("load directory: %w", err)
+ }
+
+ fmt.Printf("simulator running: %d hosted mailboxes, %d contacts (Ctrl-C to stop)\n",
+ len(s.hosted), len(s.contact))
+
+ directoryTick := time.NewTicker(60 * time.Second)
+ defer directoryTick.Stop()
+ pollTick := time.NewTicker(5 * time.Second)
+ defer pollTick.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return nil
+ case <-directoryTick.C:
+ if err := s.refreshDirectory(ctx); err != nil {
+ fmt.Printf("directory refresh failed: %v\n", err)
+ }
+ case <-pollTick.C:
+ if err := s.drainMailpit(ctx); err != nil {
+ fmt.Printf("mailpit poll failed: %v\n", err)
+ }
+ }
+ }
+}
+
+// refreshDirectory reloads who exists: hosted mailboxes (all smtp_imap
+// accounts, which the seeder pointed at the local stack) and seeded contacts.
+func (s *simulator) refreshDirectory(ctx context.Context) error {
+ hosted := map[string]hostedMailbox{}
+ rows, err := s.pool.Query(ctx, `SELECT email, name FROM email_accounts WHERE provider = 'smtp_imap'`)
+ if err != nil {
+ return err
+ }
+ for rows.Next() {
+ var m hostedMailbox
+ if err := rows.Scan(&m.Email, &m.Name); err != nil {
+ rows.Close()
+ return err
+ }
+ hosted[strings.ToLower(m.Email)] = m
+ }
+ rows.Close()
+ if err := rows.Err(); err != nil {
+ return err
+ }
+
+ contacts := map[string]contactInfo{}
+ rows, err = s.pool.Query(ctx, `SELECT email, first_name, last_name FROM contacts`)
+ if err != nil {
+ return err
+ }
+ for rows.Next() {
+ var c contactInfo
+ if err := rows.Scan(&c.Email, &c.First, &c.Last); err != nil {
+ rows.Close()
+ return err
+ }
+ contacts[strings.ToLower(c.Email)] = c
+ }
+ rows.Close()
+ if err := rows.Err(); err != nil {
+ return err
+ }
+
+ s.mu.Lock()
+ s.hosted = hosted
+ s.contact = contacts
+ s.mu.Unlock()
+ return nil
+}
+
+func (s *simulator) drainMailpit(ctx context.Context) error {
+ msgs, err := s.mp.listUnread(ctx, 200)
+ if err != nil {
+ return err
+ }
+ if len(msgs) == 0 {
+ return nil
+ }
+
+ ids := make([]string, 0, len(msgs))
+ for _, m := range msgs {
+ ids = append(ids, m.ID)
+ }
+ // Mark read up front so a crash mid-batch never double-delivers.
+ if err := s.mp.markRead(ctx, ids); err != nil {
+ return err
+ }
+
+ for _, m := range msgs {
+ if err := s.handleMessage(ctx, m); err != nil {
+ fmt.Printf("message %s (%q): %v\n", m.ID, m.Subject, err)
+ }
+ }
+ return nil
+}
+
+func (s *simulator) handleMessage(ctx context.Context, summary mailpitSummary) error {
+ detail, err := s.mp.message(ctx, summary.ID)
+ if err != nil {
+ return err
+ }
+ raw, err := s.mp.raw(ctx, summary.ID)
+ if err != nil {
+ return err
+ }
+
+ s.mu.RLock()
+ hosted := s.hosted
+ contacts := s.contact
+ s.mu.RUnlock()
+
+ for _, rcpt := range detail.To {
+ addr := strings.ToLower(rcpt.Address)
+
+ // Hosted recipient (another sandbox/warmup mailbox): final delivery
+ // into its dovecot INBOX; the worker's real IMAP sync takes it from
+ // there (warmup token verification, unibox, engagement actions).
+ if _, ok := hosted[addr]; ok {
+ if err := deliverToInbox(s.cfg.IMAPAddr, addr, s.cfg.IMAPPassword, raw); err != nil {
+ return fmt.Errorf("deliver to %s: %w", addr, err)
+ }
+ fmt.Printf("delivered %-34s %q\n", addr, detail.Subject)
+ continue
+ }
+
+ // Contact recipient: play the human.
+ if c, ok := contacts[addr]; ok {
+ sender, senderHosted := hosted[strings.ToLower(detail.From.Address)]
+ go s.actAsContact(ctx, c, detail, sender, senderHosted)
+ }
+ }
+ return nil
+}
+
+// actAsContact executes a contact persona against one received email: dwell,
+// open the pixel, maybe click, maybe reply. Runs in its own goroutine.
+func (s *simulator) actAsContact(ctx context.Context, c contactInfo, msg *mailpitMessage, sender hostedMailbox, senderHosted bool) {
+ p := personaFor(c.Email)
+ body := msg.HTML
+ if body == "" {
+ body = msg.Text
+ }
+
+ if p.Opens {
+ if task := firstMatch(pixelRe, body); task != "" {
+ s.sleep(ctx, 15*time.Second, 90*time.Second)
+ s.hitTracking(ctx, "/t/o/"+task+".png", c.Email)
+ fmt.Printf("opened %-34s %q\n", c.Email, msg.Subject)
+ }
+ }
+
+ if p.Opens && p.Clicks {
+ if ticket := firstMatch(clickRe, body); ticket != "" {
+ s.sleep(ctx, 20*time.Second, 3*time.Minute)
+ s.hitTracking(ctx, "/c/"+ticket, c.Email)
+ fmt.Printf("clicked %-34s %q\n", c.Email, msg.Subject)
+ }
+ }
+
+ if p.Opens && p.Replies && senderHosted {
+ s.sleep(ctx, 45*time.Second, 5*time.Minute)
+ text, automated := replyBody(p.Flavor, c.First)
+ reply := composeReply(
+ c.First+" "+c.Last, c.Email,
+ sender.Name, sender.Email,
+ msg.Subject, msg.MessageID, text, automated,
+ )
+ if err := deliverToInbox(s.cfg.IMAPAddr, strings.ToLower(sender.Email), s.cfg.IMAPPassword, reply); err != nil {
+ fmt.Printf("reply from %s failed: %v\n", c.Email, err)
+ return
+ }
+ fmt.Printf("replied %-34s %q\n", c.Email, msg.Subject)
+ }
+}
+
+func (s *simulator) hitTracking(ctx context.Context, path, actor string) {
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.cfg.TrackingURL+path, nil)
+ if err != nil {
+ return
+ }
+ req.Header.Set("User-Agent", userAgents[rand.Intn(len(userAgents))])
+ resp, err := s.http.Do(req)
+ if err != nil {
+ fmt.Printf("tracking hit %s for %s failed: %v\n", path, actor, err)
+ return
+ }
+ resp.Body.Close()
+}
+
+// sleep waits a humanized random duration, returning early on shutdown.
+func (s *simulator) sleep(ctx context.Context, min, max time.Duration) {
+ d := min + time.Duration(rand.Int63n(int64(max-min)))
+ select {
+ case <-ctx.Done():
+ case <-time.After(d):
+ }
+}
+
+func firstMatch(re *regexp.Regexp, body string) string {
+ m := re.FindStringSubmatch(body)
+ if len(m) < 2 {
+ return ""
+ }
+ return m[1]
+}
diff --git a/internal/seed/workers.go b/internal/seed/workers.go
index bf1c2ac1..a889a5b2 100644
--- a/internal/seed/workers.go
+++ b/internal/seed/workers.go
@@ -53,9 +53,9 @@ func seedWorkerAssignments(ctx context.Context, pool *pgxpool.Pool, _ *Result) e
// re-running the seed updates rather than duplicates.
assignmentID := uuid.MustParse("00000000-0000-0000-0000-000000000280")
_, err := pool.Exec(ctx, `
- INSERT INTO dedicated_worker_assignments (id, worker_id, user_id, subscription_id, assigned_at)
+ INSERT INTO dedicated_worker_assignments (id, worker_id, organization_id, subscription_id, assigned_at)
VALUES ($1, $2, $3, $4, NOW())
ON CONFLICT (id) DO NOTHING
- `, assignmentID, WorkerDedicatedID, UserOwnerID, SubAcmeID)
+ `, assignmentID, WorkerDedicatedID, OrgAcmeID, SubAcmeID)
return err
}
diff --git a/internal/tasks/campaign_reconciler.go b/internal/tasks/campaign_reconciler.go
index 4190f544..c616d165 100644
--- a/internal/tasks/campaign_reconciler.go
+++ b/internal/tasks/campaign_reconciler.go
@@ -15,6 +15,12 @@ import (
// re-seed. Plenty for steady state; the next tick mops up any overflow.
const campaignReconcileBatch = 500
+// overduePendingGrace is how far past its slot a pending task may run before
+// the reconcilers treat its Cloud Tasks callback as lost and cancel it. Cloud
+// Tasks fires within seconds of the scheduled time, so 15 minutes is far
+// outside any legitimate delivery or retry window.
+const overduePendingGrace = 15 * time.Minute
+
// ReconcileCampaignSchedules re-seeds the wakeup chain for active campaigns that
// have no pending task. A campaign chain is self-perpetuating (each tick
// enqueues the next), so a swallowed enqueue, a worker bounce mid-tick, or a
@@ -23,6 +29,15 @@ const campaignReconcileBatch = 500
// is the backstop that keeps them from silently stalling. Returns the number of
// chains re-seeded this pass.
func (s *tasksService) ReconcileCampaignSchedules(ctx context.Context, limit int) (int, error) {
+ // A pending task stranded well past its slot means the Cloud Tasks
+ // callback was lost (queue wipe, emulator restart, dropped retry). It
+ // blocks the "no pending task" candidate check below forever, so cancel
+ // first; a late callback for a cancelled row no-ops (handler requires
+ // 'pending').
+ if n, err := s.taskRepo.CancelOverduePendingTasks(ctx, "campaign", overduePendingGrace); err == nil && n > 0 {
+ log.Info().Int64("cancelled", n).Msg("campaign reconcile: cancelled overdue pending tasks")
+ }
+
ids, err := s.campaignRepo.ListCampaignScheduleCandidates(ctx, limit)
if err != nil {
return 0, err
diff --git a/internal/tasks/dispatch.go b/internal/tasks/dispatch.go
new file mode 100644
index 00000000..a0148466
--- /dev/null
+++ b/internal/tasks/dispatch.go
@@ -0,0 +1,47 @@
+package tasks
+
+import (
+ "context"
+ "time"
+
+ "github.com/getsentry/sentry-go"
+ "github.com/google/uuid"
+ "github.com/rs/zerolog/log"
+ "github.com/warmbly/warmbly/internal/errx"
+ "github.com/warmbly/warmbly/internal/tasks/proto"
+)
+
+// HandleTask routes a Cloud Tasks callback to the handler matching the task
+// row's task_type. All enqueues share the single CLOUD_TASKS_WEBHOOK_URL, so
+// without this dispatch a campaign task landing on the email webhook would be
+// run through the warmup handler.
+func (s *tasksService) HandleTask(task *proto.ProcessTask) *errx.Error {
+ taskID, err := uuid.Parse(task.TaskId)
+ if err != nil {
+ sentry.CaptureException(err)
+ return errx.New(errx.BadRequest, "invalid task ID")
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ rec, err := s.taskRepo.GetTask(ctx, taskID)
+ cancel()
+ if err != nil {
+ sentry.CaptureException(err)
+ return errx.InternalError()
+ }
+ if rec == nil {
+ return errx.ErrNotFound
+ }
+
+ switch rec.TaskType {
+ case "campaign":
+ return s.HandleCampaignTask(task)
+ case "warmup":
+ return s.HandleEmailTask(task)
+ case "user_email":
+ return s.HandleUserEmailTask(task)
+ default:
+ log.Warn().Str("task_id", taskID.String()).Str("task_type", rec.TaskType).Msg("task dispatch: unknown task type")
+ return errx.New(errx.BadRequest, "unknown task type")
+ }
+}
diff --git a/internal/tasks/service.go b/internal/tasks/service.go
index ae9754e8..dbde37db 100644
--- a/internal/tasks/service.go
+++ b/internal/tasks/service.go
@@ -39,6 +39,11 @@ type (
)
type TasksService interface {
+ // HandleTask routes a Cloud Tasks callback by the task row's task_type.
+ // Every enqueue shares one webhook URL (gtasks.Client holds a single
+ // callback), so the receiving endpoint cannot infer the type from the
+ // route it was hit on.
+ HandleTask(task *proto.ProcessTask) *errx.Error
HandleCampaignTask(task *proto.ProcessTask) *errx.Error
HandleEmailTask(task *proto.ProcessTask) *errx.Error
HandleUserEmailTask(task *proto.ProcessTask) *errx.Error
diff --git a/internal/tasks/warmup_reconciler.go b/internal/tasks/warmup_reconciler.go
index 25d6e9e6..25e710ab 100644
--- a/internal/tasks/warmup_reconciler.go
+++ b/internal/tasks/warmup_reconciler.go
@@ -20,6 +20,12 @@ const warmupReconcileBatch = 500
// campaign does not itself enqueue a task, so without this pass a freshly
// enabled mailbox would never start warming.
func (s *tasksService) ReconcileWarmupSchedules(ctx context.Context, limit int) (int, error) {
+ // Same lost-callback backstop as the campaign reconciler: a pending
+ // warmup task stranded past its slot blocks the candidate query below.
+ if n, err := s.taskRepo.CancelOverduePendingTasks(ctx, "warmup", overduePendingGrace); err == nil && n > 0 {
+ log.Info().Int64("cancelled", n).Msg("warmup reconcile: cancelled overdue pending tasks")
+ }
+
ids, err := s.emailRepo.ListWarmupScheduleCandidates(ctx, limit)
if err != nil {
return 0, err