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(); + }; + }, []); +}