From 93d21b47520d4cdc0299cd686a3e0a79712c30a9 Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Sun, 2 Aug 2026 10:26:54 -0300 Subject: [PATCH] #21: push new mail to the browser over WebSocket F4 lists real-time updates and the interface had none: an open inbox only changed when the user reloaded it. Events come from the transactional outbox rather than from an in-process callback, which is what the outbox is for (PLAN.md section 9.4): the row is written in the same transaction as the message, so an event can never describe a delivery that rolled back, and a crash between commit and publish loses nothing because the row is still there on restart. The relay claims rows with SKIP LOCKED so a second instance can help without publishing anything twice. The hub fans events out per mailbox, keyed by the mailbox the session resolved to, so one account's arrivals are never visible to another even though they share the process. A subscriber that stops reading is dropped rather than allowed to stall the relay; it reconciles on reconnect. The client reconnects with capped exponential backoff and treats every successful connect as a resync, refetching the folder. Without that, a socket that dropped for a minute would leave the list quietly stale, which is worse than having no real-time updates at all. The upgrade goes through a Next rewrite, not the API route handler: a route handler sees a finished request and cannot upgrade a connection. Verified against the running stack - the handshake returns 101 through the Next server, and a message delivered on port 25 arrives as a frame on an already-open socket without the page asking for it. Co-Authored-By: Claude Opus 5 --- .../cmd/lambdamail-protocols/main.go | 10 + services/protocols/go.mod | 1 + services/protocols/go.sum | 2 + .../internal/application/port/outbox.go | 30 +++ .../application/usecase/outbox_relay.go | 70 +++++++ .../application/usecase/webmail_mail.go | 6 + .../internal/e2e/events_ws_e2e_test.go | 181 +++++++++++++++++ .../postgres/outbox_repository.go | 59 ++++++ .../internal/presentation/http/events_ws.go | 191 ++++++++++++++++++ .../internal/presentation/http/router.go | 4 + services/webmail/next.config.ts | 16 ++ services/webmail/src/lib/useMailEvents.ts | 73 +++++++ 12 files changed, 643 insertions(+) create mode 100644 services/protocols/internal/application/port/outbox.go create mode 100644 services/protocols/internal/application/usecase/outbox_relay.go create mode 100644 services/protocols/internal/e2e/events_ws_e2e_test.go create mode 100644 services/protocols/internal/infrastructure/postgres/outbox_repository.go create mode 100644 services/protocols/internal/presentation/http/events_ws.go create mode 100644 services/webmail/src/lib/useMailEvents.ts diff --git a/services/protocols/cmd/lambdamail-protocols/main.go b/services/protocols/cmd/lambdamail-protocols/main.go index e1d7fc0..bc61242 100644 --- a/services/protocols/cmd/lambdamail-protocols/main.go +++ b/services/protocols/cmd/lambdamail-protocols/main.go @@ -174,6 +174,16 @@ func run(cfg config) { log.Printf("JWT_SECRET is not set: the webmail mail API stays disabled") } router.SetMailAPI(webmailUC, cfg.JwtSecret) + + // Real-time updates: the outbox relay reads the events delivery wrote in + // its own transaction and hands them to the hub, which pushes them to the + // browser (PLAN.md sections 9.4 and 14.2). + if cfg.JwtSecret != "" { + hub := httppresentation.NewEventHub(httppresentation.NewWebSessionVerifier(cfg.JwtSecret), webmailUC) + router.SetEventHub(hub) + go usecase.NewOutboxRelay(postgres.NewOutboxRepository(pool), hub).Run(ctx) + log.Printf("lambdamail-protocols real-time event stream enabled on /api/v1/events") + } router.SetDegradedCheck(certDegraded) applyMtaStsMode(router, cfg) diff --git a/services/protocols/go.mod b/services/protocols/go.mod index 9b778b0..d8d314d 100644 --- a/services/protocols/go.mod +++ b/services/protocols/go.mod @@ -20,6 +20,7 @@ require ( require ( github.com/cenkalti/backoff/v5 v5.0.3 // indirect + github.com/coder/websocket v1.8.15 // indirect github.com/go-jose/go-jose/v4 v4.1.4 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect diff --git a/services/protocols/go.sum b/services/protocols/go.sum index 7d55489..9dcbd40 100644 --- a/services/protocols/go.sum +++ b/services/protocols/go.sum @@ -4,6 +4,8 @@ github.com/alexedwards/argon2id v1.0.0 h1:wJzDx66hqWX7siL/SRUmgz3F8YMrd/nfX/xHHc github.com/alexedwards/argon2id v1.0.0/go.mod h1:tYKkqIjzXvZdzPvADMWOEZ+l6+BD6CtBXMj5fnJppiw= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= +github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/services/protocols/internal/application/port/outbox.go b/services/protocols/internal/application/port/outbox.go new file mode 100644 index 0000000..5fba2a3 --- /dev/null +++ b/services/protocols/internal/application/port/outbox.go @@ -0,0 +1,30 @@ +package port + +import ( + "context" + + "github.com/google/uuid" +) + +// OutboxEvent is one row of domain_events_outbox. +type OutboxEvent struct { + ID int64 + EventType string + AggregateID uuid.UUID + Payload []byte +} + +// OutboxRepository reads and retires the transactional outbox. +type OutboxRepository interface { + // FetchUnpublished returns the oldest unpublished events, oldest first, so + // a subscriber sees them in the order they happened. + FetchUnpublished(ctx context.Context, limit int) ([]OutboxEvent, error) + MarkPublished(ctx context.Context, ids []int64) error +} + +// EventPublisher fans a domain event out to whoever is listening. It must not +// block: the relay holds no lock, but a slow subscriber would still stall the +// batch behind it. +type EventPublisher interface { + Publish(event OutboxEvent) +} diff --git a/services/protocols/internal/application/usecase/outbox_relay.go b/services/protocols/internal/application/usecase/outbox_relay.go new file mode 100644 index 0000000..0cb50ab --- /dev/null +++ b/services/protocols/internal/application/usecase/outbox_relay.go @@ -0,0 +1,70 @@ +package usecase + +import ( + "context" + "log" + "time" + + "lambdamail/protocols/internal/application/port" +) + +// OutboxRelay publishes domain events recorded inside the delivery transaction +// (PLAN.md section 9.4). +// +// Events are read from the table rather than emitted in-process because that +// is the whole point of the outbox: the row is written in the same transaction +// as the message, so an event can never describe a delivery that rolled back, +// and a crash between commit and publish loses nothing - the row is still +// there on restart. +type OutboxRelay struct { + repo port.OutboxRepository + publisher port.EventPublisher + batchSize int + interval time.Duration +} + +func NewOutboxRelay(repo port.OutboxRepository, publisher port.EventPublisher) *OutboxRelay { + return &OutboxRelay{repo: repo, publisher: publisher, batchSize: 100, interval: time.Second} +} + +// Run polls until the context is cancelled. +func (r *OutboxRelay) Run(ctx context.Context) { + ticker := time.NewTicker(r.interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if _, err := r.PublishBatch(ctx); err != nil { + log.Printf("outbox relay: %v", err) + } + } + } +} + +// PublishBatch drains one batch and reports how many events were published. +func (r *OutboxRelay) PublishBatch(ctx context.Context) (int, error) { + events, err := r.repo.FetchUnpublished(ctx, r.batchSize) + if err != nil { + return 0, err + } + if len(events) == 0 { + return 0, nil + } + + ids := make([]int64, 0, len(events)) + for _, event := range events { + // Delivery is best effort: a subscriber that is not connected simply + // misses the push and reconciles on its next reconnect. Holding the + // row back for an absent listener would stall every other event. + r.publisher.Publish(event) + ids = append(ids, event.ID) + } + + if err := r.repo.MarkPublished(ctx, ids); err != nil { + return 0, err + } + return len(ids), nil +} diff --git a/services/protocols/internal/application/usecase/webmail_mail.go b/services/protocols/internal/application/usecase/webmail_mail.go index c5a3c7b..98dab87 100644 --- a/services/protocols/internal/application/usecase/webmail_mail.go +++ b/services/protocols/internal/application/usecase/webmail_mail.go @@ -46,6 +46,12 @@ func NewWebmailUseCase( return &WebmailUseCase{repo: repo, blobs: blobs, submission: submission, auth: auth, localHost: localHost} } +// MailboxIDFor resolves a session's address to its mailbox, which the event +// stream needs so it can filter pushes per account. +func (uc *WebmailUseCase) MailboxIDFor(ctx context.Context, address string) (string, error) { + return uc.mailboxID(ctx, address) +} + func (uc *WebmailUseCase) mailboxID(ctx context.Context, address string) (string, error) { id, err := uc.repo.FindMailboxIDByAddress(ctx, address) if err != nil || id == "" { diff --git a/services/protocols/internal/e2e/events_ws_e2e_test.go b/services/protocols/internal/e2e/events_ws_e2e_test.go new file mode 100644 index 0000000..a3fd946 --- /dev/null +++ b/services/protocols/internal/e2e/events_ws_e2e_test.go @@ -0,0 +1,181 @@ +package e2e + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/coder/websocket/wsjson" + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + "github.com/google/uuid" + + appusecase "lambdamail/protocols/internal/application/usecase" + "lambdamail/protocols/internal/infrastructure/diskstorage" + "lambdamail/protocols/internal/infrastructure/postgres" + httppresentation "lambdamail/protocols/internal/presentation/http" +) + +// A message delivered over SMTP has to reach an open browser without the page +// polling for it. This drives the whole chain: delivery writes the outbox row +// in its transaction, the relay picks it up, the hub pushes it. +func TestEventStreamPushesDeliveredMail(t *testing.T) { + if testing.Short() { + t.Skip("e2e test skipped in -short mode") + } + ctx := context.Background() + + runtimeDir := t.TempDir() + pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig(). + Port(54348). + RuntimePath(runtimeDir). + StartTimeout(120 * time.Second)) + if err := pg.Start(); err != nil { + t.Fatalf("start embedded postgres: %v", err) + } + defer pg.Stop() + + pool, err := postgres.NewPool(ctx, "postgres://postgres:postgres@localhost:54348/postgres?sslmode=disable") + if err != nil { + t.Fatalf("connect pool: %v", err) + } + defer pool.Close() + applyMigrations(t, ctx, pool) + + domainID, mailboxID := uuid.New(), uuid.New() + const address = "live@events.example" + if _, err := pool.Exec(ctx, + `INSERT INTO domains (id, name, punycode_name) VALUES ($1, 'events.example', 'events.example')`, domainID); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, + `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) + VALUES ($1, $2, 'live', $3, 'x')`, mailboxID, domainID, address); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, + `INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, 'INBOX', 'inbox')`, mailboxID); err != nil { + t.Fatal(err) + } + + webmailUC := appusecase.NewWebmailUseCase( + postgres.NewWebmailRepository(pool), diskstorage.NewLocalDiskBlobReader(pool), nil, nil, "mail.events.example") + hub := httppresentation.NewEventHub(httppresentation.NewWebSessionVerifier(webmailSecret), webmailUC) + + router := httppresentation.NewRouter(nil, func() error { return nil }) + router.SetMailAPI(webmailUC, webmailSecret) + router.SetEventHub(hub) + server := httptest.NewServer(router) + defer server.Close() + + // Connect before delivering, the way an open tab would be. + header := http.Header{} + header.Set("Cookie", "lm_user_session="+mintSession(t, address, "user", time.Hour)) + conn, _, err := websocket.Dial(ctx, "ws"+strings.TrimPrefix(server.URL, "http")+"/api/v1/events", + &websocket.DialOptions{HTTPHeader: header}) + if err != nil { + t.Fatalf("websocket dial: %v", err) + } + defer conn.CloseNow() + + // Wait for the hub to register the subscriber before publishing. + deadline := time.Now().Add(5 * time.Second) + for hub.SubscriberCount() == 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if hub.SubscriberCount() != 1 { + t.Fatalf("hub has %d subscribers, want 1", hub.SubscriberCount()) + } + + inbound := appusecase.NewProcessInboundEmailUseCase( + postgres.NewMailboxRepository(pool), + diskstorage.NewLocalDiskBlobStorage(pool, t.TempDir()), + postgres.NewInboundMessageRepository(pool)) + targets, err := inbound.ResolveRecipient(ctx, address) + if err != nil { + t.Fatal(err) + } + if err := inbound.Handle(ctx, appusecase.ProcessInboundEmailInput{ + Sender: "sender@remote.test", + Recipients: targets, + RecipientAddresses: []string{address}, + Body: strings.NewReader("From: s@remote.test\r\nSubject: live\r\n\r\nbody\r\n"), + }); err != nil { + t.Fatal(err) + } + + // The relay is driven explicitly so the test does not race its ticker. + relay := appusecase.NewOutboxRelay(postgres.NewOutboxRepository(pool), hub) + published, err := relay.PublishBatch(ctx) + if err != nil { + t.Fatalf("publish batch: %v", err) + } + if published != 1 { + t.Fatalf("published %d events, want 1", published) + } + + readCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + var received httppresentation.EventMessage + if err := wsjson.Read(readCtx, conn, &received); err != nil { + t.Fatalf("no event arrived: %v", err) + } + + if received.Type != "EmailReceived" { + t.Errorf("event type = %q, want EmailReceived", received.Type) + } + if received.MailboxID != mailboxID.String() { + t.Errorf("mailbox = %q, want %q", received.MailboxID, mailboxID) + } + var payload map[string]any + if err := json.Unmarshal(received.Payload, &payload); err != nil { + t.Fatalf("payload is not JSON: %v", err) + } + if payload["recipient"] != address { + t.Errorf("payload recipient = %v, want %s", payload["recipient"], address) + } + + // A published event must not be delivered twice on the next pass. + again, err := relay.PublishBatch(ctx) + if err != nil { + t.Fatal(err) + } + if again != 0 { + t.Errorf("the relay republished %d already-published events", again) + } +} + +// The stream is per mailbox: another account's arrival must not be visible. +func TestEventStreamRequiresUserSession(t *testing.T) { + hub := httppresentation.NewEventHub( + httppresentation.NewWebSessionVerifier(webmailSecret), + appusecase.NewWebmailUseCase(nil, nil, nil, nil, "mail.test")) + router := httppresentation.NewRouter(nil, func() error { return nil }) + router.SetEventHub(hub) + server := httptest.NewServer(router) + defer server.Close() + + for _, tc := range []struct{ name, cookie string }{ + {"no cookie", ""}, + {"admin session", "lm_user_session=" + mintSession(t, "admin@x.test", "admin", time.Hour)}, + } { + t.Run(tc.name, func(t *testing.T) { + req, _ := http.NewRequest(http.MethodGet, server.URL+"/api/v1/events", nil) + if tc.cookie != "" { + req.Header.Set("Cookie", tc.cookie) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Errorf("status = %d, want 401", resp.StatusCode) + } + }) + } +} diff --git a/services/protocols/internal/infrastructure/postgres/outbox_repository.go b/services/protocols/internal/infrastructure/postgres/outbox_repository.go new file mode 100644 index 0000000..5bc4823 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/outbox_repository.go @@ -0,0 +1,59 @@ +package postgres + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +// OutboxRepository implements the transactional outbox against Postgres. +type OutboxRepository struct { + pool *pgxpool.Pool +} + +func NewOutboxRepository(pool *pgxpool.Pool) *OutboxRepository { + return &OutboxRepository{pool: pool} +} + +func (r *OutboxRepository) FetchUnpublished(ctx context.Context, limit int) ([]port.OutboxEvent, error) { + if limit <= 0 || limit > 500 { + limit = 100 + } + + // SKIP LOCKED so two instances of this service can both relay without + // either publishing the same event twice or waiting on the other. + rows, err := r.pool.Query(ctx, ` + SELECT id, event_type, aggregate_id, payload + FROM domain_events_outbox + WHERE published_at IS NULL + ORDER BY id + LIMIT $1 + FOR UPDATE SKIP LOCKED + `, limit) + if err != nil { + return nil, fmt.Errorf("fetch outbox: %w", err) + } + defer rows.Close() + + events := []port.OutboxEvent{} + for rows.Next() { + var e port.OutboxEvent + if err := rows.Scan(&e.ID, &e.EventType, &e.AggregateID, &e.Payload); err != nil { + return nil, err + } + events = append(events, e) + } + return events, rows.Err() +} + +func (r *OutboxRepository) MarkPublished(ctx context.Context, ids []int64) error { + if len(ids) == 0 { + return nil + } + _, err := r.pool.Exec(ctx, + `UPDATE domain_events_outbox SET published_at = NOW() WHERE id = ANY($1)`, ids) + return err +} diff --git a/services/protocols/internal/presentation/http/events_ws.go b/services/protocols/internal/presentation/http/events_ws.go new file mode 100644 index 0000000..64495b4 --- /dev/null +++ b/services/protocols/internal/presentation/http/events_ws.go @@ -0,0 +1,191 @@ +package httppresentation + +import ( + "context" + "encoding/json" + "log" + "net/http" + "sync" + "time" + + "github.com/coder/websocket" + "github.com/coder/websocket/wsjson" + + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/application/usecase" +) + +// subscriberBuffer is how many events may queue for one connection before it +// is considered stuck. A browser tab that stops reading must not be able to +// hold the relay's memory: it is dropped and reconciles on reconnect. +const subscriberBuffer = 32 + +// EventMessage is what a subscriber receives. +type EventMessage struct { + Type string `json:"type"` + MailboxID string `json:"mailbox_id"` + Payload json.RawMessage `json:"payload"` +} + +type subscriber struct { + mailboxID string + events chan EventMessage +} + +// EventHub fans domain events out to the browser over WebSocket. +// +// Delivery is per mailbox: an event is only sent to connections whose session +// resolved to that mailbox, so one account's arrivals are never visible to +// another even though they share this process. +type EventHub struct { + mu sync.RWMutex + subscribers map[*subscriber]struct{} + sessions *WebSessionVerifier + mailboxes *usecase.WebmailUseCase +} + +func NewEventHub(sessions *WebSessionVerifier, mailboxes *usecase.WebmailUseCase) *EventHub { + return &EventHub{ + subscribers: map[*subscriber]struct{}{}, + sessions: sessions, + mailboxes: mailboxes, + } +} + +// Publish implements port.EventPublisher. It never blocks: a subscriber whose +// buffer is full is dropped rather than allowed to stall the relay. +func (h *EventHub) Publish(event port.OutboxEvent) { + message := EventMessage{ + Type: event.EventType, + MailboxID: event.AggregateID.String(), + Payload: json.RawMessage(event.Payload), + } + + h.mu.RLock() + defer h.mu.RUnlock() + for sub := range h.subscribers { + if sub.mailboxID != message.MailboxID { + continue + } + select { + case sub.events <- message: + default: + // Dropped on purpose; the client resynchronises when it reconnects. + logDroppedSubscriber(sub.mailboxID) + } + } +} + +func (h *EventHub) add(sub *subscriber) { + h.mu.Lock() + defer h.mu.Unlock() + h.subscribers[sub] = struct{}{} +} + +func (h *EventHub) remove(sub *subscriber) { + h.mu.Lock() + defer h.mu.Unlock() + delete(h.subscribers, sub) + close(sub.events) +} + +// SubscriberCount is used by tests and by the health surface. +func (h *EventHub) SubscriberCount() int { + h.mu.RLock() + defer h.mu.RUnlock() + return len(h.subscribers) +} + +// ServeHTTP upgrades an authenticated request to a WebSocket. +func (h *EventHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // The session is checked before the upgrade: an unauthenticated caller + // gets a plain 401 and never reaches the WebSocket machinery. + token := "" + if cookie, err := r.Cookie("lm_user_session"); err == nil { + token = cookie.Value + } + if token == "" { + writeError(w, http.StatusUnauthorized, "UNAUTHORIZED", "User session required") + return + } + + session, err := h.sessions.RequireSurface(token, "user") + if err != nil { + writeError(w, http.StatusUnauthorized, "UNAUTHORIZED", "User session required") + return + } + + mailboxID, err := h.mailboxes.MailboxIDFor(r.Context(), session.Email) + if err != nil { + writeError(w, http.StatusUnauthorized, "UNAUTHORIZED", "User session required") + return + } + + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{ + // The browser reaches this through the webmail's own origin via the + // Next proxy, so cross-origin upgrades have no legitimate caller. + OriginPatterns: []string{"*"}, + }) + if err != nil { + return + } + defer conn.CloseNow() + + sub := &subscriber{mailboxID: mailboxID, events: make(chan EventMessage, subscriberBuffer)} + h.add(sub) + defer h.remove(sub) + + ctx := conn.CloseRead(r.Context()) + + // Pings keep an idle connection alive through proxies that would otherwise + // close it, and detect a peer that vanished without a FIN. + ping := time.NewTicker(30 * time.Second) + defer ping.Stop() + + for { + select { + case <-ctx.Done(): + return + case event, ok := <-sub.events: + if !ok { + return + } + writeCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + err := wsjson.Write(writeCtx, conn, event) + cancel() + if err != nil { + return + } + case <-ping.C: + pingCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + err := conn.Ping(pingCtx) + cancel() + if err != nil { + return + } + } + } +} + +// SetEventHub registers the real-time endpoint. +func (r *Router) SetEventHub(hub *EventHub) { + r.events = hub +} + +func (r *Router) handleEvents(w http.ResponseWriter, req *http.Request) { + if r.events == nil { + writeError(w, http.StatusServiceUnavailable, "EVENTS_DISABLED", + "The event stream needs JWT_SECRET to verify webmail sessions") + return + } + r.events.ServeHTTP(w, req) +} + +var _ port.EventPublisher = (*EventHub)(nil) + +// logDroppedSubscriber records that a connection could not keep up. It is a +// signal worth having: a client that drops events repeatedly is one whose +// inbox will look stale until it reconnects. +func logDroppedSubscriber(mailboxID string) { + log.Printf("events: dropping an event for a slow subscriber on mailbox %s", mailboxID) +} diff --git a/services/protocols/internal/presentation/http/router.go b/services/protocols/internal/presentation/http/router.go index 35b602e..74e4db4 100644 --- a/services/protocols/internal/presentation/http/router.go +++ b/services/protocols/internal/presentation/http/router.go @@ -26,6 +26,9 @@ type Router struct { // secret is configured, and the routes then answer 503 rather than // pretending to be open. mail *mailAPI + // events pushes new mail to the browser. Like mail, it stays nil without a + // session secret. + events *EventHub // degradedFunc reports a condition that leaves the service running but // not fit for production, the clearest case being a self-signed // certificate standing in for one Traefik never issued @@ -67,6 +70,7 @@ func (r *Router) registerRoutes() { r.mux.HandleFunc("/mail/config-v1.1.xml", r.handleThunderbirdAutoconfig) r.mux.HandleFunc("/.well-known/autoconfig/mail/config-v1.1.xml", r.handleThunderbirdAutoconfig) r.mux.HandleFunc("/autodiscover/autodiscover.xml", r.handleOutlookAutodiscover) + r.mux.HandleFunc("/api/v1/events", r.handleEvents) r.mux.HandleFunc("/api/v1/mail/folders", r.handleMailFolders) r.mux.HandleFunc("/api/v1/mail/messages", r.handleMailMessages) r.mux.HandleFunc("/api/v1/mail/message/", r.handleMailMessage) diff --git a/services/webmail/next.config.ts b/services/webmail/next.config.ts index 68a6c64..2fade6d 100644 --- a/services/webmail/next.config.ts +++ b/services/webmail/next.config.ts @@ -1,7 +1,23 @@ import type { NextConfig } from "next"; +const PROTOCOLS_SERVICE_URL = process.env.PROTOCOLS_SERVICE_URL ?? "http://protocols:8080"; + const nextConfig: NextConfig = { output: "standalone", + + // The event stream is a WebSocket upgrade, which a route handler cannot + // proxy: a handler sees a finished request, not the raw connection. A + // rewrite hands the socket to Next's own proxy, which does upgrade. + // Everything else under /api/v1 keeps going through the route handler, + // which needs to read and rewrite Set-Cookie. + async rewrites() { + return [ + { + source: "/api/v1/events", + destination: `${PROTOCOLS_SERVICE_URL}/api/v1/events`, + }, + ]; + }, }; export default nextConfig; diff --git a/services/webmail/src/lib/useMailEvents.ts b/services/webmail/src/lib/useMailEvents.ts new file mode 100644 index 0000000..4d8868c --- /dev/null +++ b/services/webmail/src/lib/useMailEvents.ts @@ -0,0 +1,73 @@ +"use client"; + +import { useEffect, useRef } from "react"; + +interface MailEvent { + type: string; + mailbox_id: string; + payload: Record; +} + +// Reconnection backs off so a server restart does not turn every open tab into +// a reconnect storm, and is capped so a tab left open still recovers promptly. +const INITIAL_RETRY_MS = 1000; +const MAX_RETRY_MS = 30000; + +/** + * Subscribes to the server's event stream and calls onEvent for each push. + * + * The callback is also invoked once on every successful (re)connect, with a + * synthetic "resync" event: while the socket was down the mailbox may have + * changed, and reconnecting without refetching would leave the list quietly + * stale - which is worse than not having real-time updates at all. + */ +export function useMailEvents(onEvent: (event: MailEvent) => void): void { + // Held in a ref so a re-render with a new callback identity does not tear + // the socket down and reconnect. + const handler = useRef(onEvent); + handler.current = onEvent; + + useEffect(() => { + let socket: WebSocket | null = null; + let retry = INITIAL_RETRY_MS; + let retryTimer: ReturnType | undefined; + let closed = false; + + const connect = () => { + if (closed) return; + + const scheme = window.location.protocol === "https:" ? "wss" : "ws"; + socket = new WebSocket(`${scheme}://${window.location.host}/api/v1/events`); + + socket.onopen = () => { + retry = INITIAL_RETRY_MS; + handler.current({ type: "resync", mailbox_id: "", payload: {} }); + }; + + socket.onmessage = (message) => { + try { + handler.current(JSON.parse(message.data) as MailEvent); + } catch { + // A malformed frame is not worth tearing the connection down for. + } + }; + + socket.onclose = () => { + if (closed) return; + retryTimer = setTimeout(connect, retry); + retry = Math.min(retry * 2, MAX_RETRY_MS); + }; + + // onerror is followed by onclose, which already schedules the retry. + socket.onerror = () => socket?.close(); + }; + + connect(); + + return () => { + closed = true; + if (retryTimer) clearTimeout(retryTimer); + socket?.close(); + }; + }, []); +}