Deploy

origin := os.Getenv("VIA_ORIGIN")
if origin == "" {
	return errors.New("VIA_ORIGIN unset")
}
// No WithSessionKey: via reads VIA_SESSION_KEY from the environment.
r := via.NewRouter(
	via.WithTrustedOrigin(origin),
	via.WithSecureCookies(),
	via.WithSessionStore(SQLSessions{DB: db}),
	via.WithLogger(slog.New(slog.NewJSONHandler(os.Stderr, nil))),
	via.WithMaxSSEConn(5000),
)
via.Mount(r, "/", Home{})

example.com {
	# text/* would also compress, and so buffer, text/event-stream.
	encode zstd gzip {
		match {
			header Content-Type text/html*
			header Content-Type text/css*
			header Content-Type text/plain*
			header Content-Type text/javascript*
			header Content-Type application/javascript*
			header Content-Type application/json*
			header Content-Type image/svg+xml*
		}
	}

	reverse_proxy 127.0.0.1:8080 {
		# SSE: never buffer the upstream stream.
		flush_interval -1
	}
}

# In the http block: open requests per client address, each open stream
# included. Clients behind one NAT share an address, so leave headroom.
limit_conn_zone $binary_remote_addr zone=perclient:10m;

server {
	listen 443 ssl;
	http2 on;
	server_name example.com;
	ssl_certificate     /etc/ssl/example.com/fullchain.pem;
	ssl_certificate_key /etc/ssl/example.com/privkey.pem;

	location / {
		limit_conn perclient 50;
		proxy_pass http://127.0.0.1:8080;
		proxy_http_version 1.1;
		proxy_set_header Connection "";
		proxy_set_header Host $host;
		proxy_set_header X-Forwarded-Proto $scheme;
		# SSE: pass each frame on as it arrives.
		proxy_buffering off;
		# Must exceed via's fixed 25 s keepalive.
		proxy_read_timeout 60s;
	}
}

srv := &http.Server{
	Addr:              cmp.Or(os.Getenv("VIA_ADDR"), "127.0.0.1:8080"),
	Handler:           mux,
	ReadHeaderTimeout: 10 * time.Second,
	ReadTimeout:       30 * time.Second,
	IdleTimeout:       120 * time.Second,
	// WriteTimeout stays 0: it bounds the whole response, and a stream is
	// one response for the life of the tab.
}

var ready atomic.Bool
mux := http.NewServeMux()
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
	io.WriteString(w, "ok\n")
})
mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, _ *http.Request) {
	if !ready.Load() {
		http.Error(w, "draining", http.StatusServiceUnavailable)
		return
	}
	io.WriteString(w, "ok\n")
})
mux.Handle("/", r)

serve := make(chan error, 1)
go func() { serve <- srv.ListenAndServe() }()
ready.Store(true)

select {
case err := <-serve:
	return err
case <-ctx.Done():
}
stop()

ready.Store(false)
time.Sleep(drainDelay)
shut, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
return errors.Join(r.Shutdown(shut), srv.Shutdown(shut))

[Unit]
Description=myapp
After=network-online.target
Wants=network-online.target
StartLimitIntervalSec=0

[Service]
ExecStart=/usr/local/bin/myapp
Environment=VIA_ADDR=127.0.0.1:8080
Environment=VIA_ORIGIN=https://example.com
EnvironmentFile=/etc/myapp.env
User=myapp
Restart=always
RestartSec=2
TimeoutStopSec=15
NoNewPrivileges=yes
CapabilityBoundingSet=
UMask=0077
ProtectSystem=strict
ProtectHome=yes
PrivateTmp=yes
PrivateDevices=yes
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
SystemCallFilter=@system-service
SystemCallArchitectures=native

[Install]
WantedBy=multi-user.target

#!/sbin/openrc-run
description="myapp"

supervisor=supervise-daemon
command=/usr/local/bin/myapp
command_user=myapp:myapp
respawn_delay=2
respawn_max=0
output_log=/var/log/myapp.log
error_log=/var/log/myapp.log

depend() {
	need net
	before caddy
}

start_pre() {
	set -a
	. /etc/myapp.env
	set +a
	export VIA_ADDR=127.0.0.1:8080
	export VIA_ORIGIN=https://example.com
	checkpath -f -o myapp:myapp -m 0600 /var/log/myapp.log
}

// SQLSessions keeps sessions in one Postgres table:
//
//	CREATE TABLE via_sessions (
//		id      text PRIMARY KEY,
//		data    bytea NOT NULL,
//		expires timestamptz NOT NULL
//	);
type SQLSessions struct{ DB *sql.DB }

func (s SQLSessions) Load(ctx context.Context, id string) ([]byte, bool, error) {
	var data []byte
	err := s.DB.QueryRowContext(ctx, `
		SELECT data FROM via_sessions
		WHERE id = $1 AND expires > now()`, id).Scan(&data)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, false, nil
	}
	return data, err == nil, err
}

func (s SQLSessions) Save(ctx context.Context, id string, data []byte, ttl time.Duration) error {
	_, err := s.DB.ExecContext(ctx, `
		INSERT INTO via_sessions (id, data, expires) VALUES ($1, $2, $3)
		ON CONFLICT (id) DO UPDATE
		SET data = excluded.data, expires = excluded.expires`,
		id, data, time.Now().Add(ttl))
	return err
}

func (s SQLSessions) Delete(ctx context.Context, id string) error {
	_, err := s.DB.ExecContext(ctx, `DELETE FROM via_sessions WHERE id = $1`, id)
	return err
}

// Bus is the part of a pub/sub client the bridge needs.
type Bus interface {
	Publish(ctx context.Context, subject string, payload []byte) error
	Subscribe(ctx context.Context, subject string) (<-chan []byte, error)
}

type Room struct {
	ID    string
	Bus   Bus
	Local *topic.Topic[Message]
}

// Bridge feeds what this pod hears on the bus into its local topic. Run one
// per room per pod, for the life of the process.
func (r *Room) Bridge(ctx context.Context) error {
	in, err := r.Bus.Subscribe(ctx, "room."+r.ID)
	if err != nil {
		return err
	}
	for b := range in {
		var m Message
		if json.Unmarshal(b, &m) == nil {
			r.Local.Publish(m)
		}
	}
	return ctx.Err()
}

// Post publishes outward, never to r.Local: the sender's own pod hears it back
// through Bridge like every other pod does.
func (r *Room) Post(ctx context.Context, m Message) error {
	b, err := json.Marshal(m)
	if err != nil {
		return err
	}
	return r.Bus.Publish(ctx, "room."+r.ID, b)
}

type Chat struct {
	Room *Room
	Log  via.List[Message]
}

func (c *Chat) OnInit(ctx *via.Ctx) error {
	ctx.Listen(c.Room.Local, c.onMessage)
	return nil
}

func (c *Chat) onMessage(ctx *via.Ctx, m Message) { c.Log.Append(m) }