Documentation
¶
Index ¶
Constants ¶
const ( ReasonReplaced = "replaced" ReasonShuttingDown = "shutting_down" ReasonPingTimeout = "ping_timeout" )
Stable close-reason tokens (the machine-readable part of the contract).
const ( // StatusReplaced (4000) — this connection was superseded by a NEWER // connection for the same agent (one-connection-per-agent policy). This is // benign from the server's perspective, but the superseded client must NOT // auto-reconnect: reconnecting would steal the socket back from its // replacement and loop. StatusReplaced websocket.StatusCode = 4000 )
WebSocket close-code contract (GA-frozen; documented in docs/api.md "Real-time delivery (WebSocket)" → "Connection lifecycle & close codes").
The server enforces one connection per agent: a newer connection for the same agent supersedes the current one. Before this contract, the replaced path reused 1008 (policy violation) — the same code a genuine rejection would use — so an SDK could not tell "your own newer connection took over (stop)" from "the server refuses you (stop)" from a transient drop (reconnect), and two of a user's own processes would steal the socket from each other in a reconnect loop.
code | reason token | meaning | client action
-----+----------------+-------------------------------------------+---------------------------
1000 | (empty) | normal closure | none (don't reconnect)
1001 | shutting_down | server shutdown/restart (deploy) | reconnect with backoff
1001 | ping_timeout | server dropped an unresponsive connection | reconnect with backoff
1008 | (message) | genuine policy rejection after upgrade | stop; do NOT reconnect
1011 | (message) | internal server error | reconnect with backoff
4000 | replaced | a newer connection for this agent | stop; do NOT reconnect
| | superseded this one |
4001–4999 reserved for future e2a-specific fatal | stop; do NOT reconnect
conditions (e.g. agent deleted) |
Reason strings on e2a-specific closes are short stable snake_case tokens ("replaced", "shutting_down", "ping_timeout") — safe to branch on, though clients should branch on the CODE first. Handshake rejections (bad key, wrong agent, cross-tenant) happen before the upgrade and are HTTP error envelopes (401/403/404), never close codes.
Variables ¶
This section is empty.
Functions ¶
func BuildNotification ¶
BuildNotification reconstructs the same versioned envelope SHAPE for stores that cannot return the durable event — {type:"email.received", id, schema_version, created_at, data} with the canonical typed eventpayload.EmailReceivedData — so a consumer can share one parser across both channels. The event id uses the same deterministic derivation the outbox uses (sha256(message_id|type)), so a subscriber that receives the message on both channels can dedup on it.
This drain-path rebuild populates what the message row provides, including the canonical SMTP authentication evidence persisted at intake. Consumers therefore get the same DMARC verdict whether the frame arrives live or on a reconnect drain.
Production reconnect drain uses NotificationForMessage and therefore reuses the exact persisted envelope. This builder remains the compatibility fallback for stores that do not expose durable events.
The live relay path (internal/relay) marshals the actual outbox event: the same event envelope — identical fields and event id — as the webhook delivery (byte layout may differ across channels; JSON key order/escaping is not contractual).
func NotificationForMessage ¶
NotificationForMessage prefers the immutable envelope persisted by the transactional outbox. The fallback preserves compatibility for test/legacy stores without the outbox lookup seam.
Types ¶
type Handler ¶
type Handler struct {
// contains filtered or unexported fields
}
Handler upgrades HTTP connections to WebSocket. Open to any agent the caller owns — the legacy local-mode gate was removed (migration 029).
func NewHandler ¶
func NewHandler(hub *Hub, store HandlerStore) *Handler
func (*Handler) ServeWithEmail ¶
ServeWithEmail handles the WebSocket upgrade for an explicitly-provided agent email — the router owns path extraction (and, for routers like chi that return params still percent-encoded, DECODING; see the /v1 mount in internal/httpapi). The old gorilla/mux `Handle` variant was removed with the retired /api/v1 surface: this is the transport's only entry point.
type HandlerStore ¶
type HandlerStore interface {
// GetPrincipalByAPIKey returns the scoped principal (User + Scope + AgentID)
// so the WS transport can enforce the SAME agent-scope confinement the REST
// API does (HIGH-1): an agent-scoped key must not open another agent's stream.
GetPrincipalByAPIKey(ctx context.Context, apiKey string) (*identity.Principal, error)
GetAgentByEmail(ctx context.Context, email string) (*identity.AgentIdentity, error)
GetMessagesByAgent(ctx context.Context, f identity.MessageListFilter) ([]identity.Message, error)
}
HandlerStore is the subset of identity.Store that the WS handler needs.
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub manages WebSocket connections for agents that live-tail inbound mail. Any agent may connect — WS is an opportunistic push on top of the durable pollable inbox + webhook subscriptions. One connection per agent; new connections replace old ones.
func (*Hub) IsConnected ¶
IsConnected returns whether an agent has an active WebSocket connection.
func (*Hub) Register ¶
Register stores a connection for the given agent, returning any previous connection. The caller should close the old connection if non-nil.
Source Files