Real-time Streaming

Alphorn supports real-time notification streaming via Server-Sent Events (SSE). Build live dashboards, integrate with custom UIs, or connect to any system that supports SSE.

How It WorksCopy Anchor Link

The Live Stream (SSE) channel is built into Alphorn. Clients subscribe to the channel's stream URL and receive notifications as they arrive - no polling required.

1. Create a channel of type Live Stream (SSE) under Channels
2. Choose the output format and which fields to include (see Event Format)
3. Link the channel to one or more webhooks, like any other channel
4. Copy the Stream URL from the channel edit page - it looks like https://app.alphorn.dev/api/stream/<stream-id>

The stream ID belongs to the channel, not to a webhook. Every webhook linked to the channel (and passing its filter) shows up on the same stream.

Streaming is live only: notifications are pushed to the clients connected at that moment. There is no replay of missed events - events carry no id: field, so Last-Event-ID resuming isn't possible and a client that was offline will not receive what it missed. The delivery is marked as delivered even when no client is connected. Live Stream channels have no test button.

Connecting to a StreamCopy Anchor Link

JavaScript (browser)Copy Anchor Link

const eventSource = new EventSource(
  "https://app.alphorn.dev/api/stream/0199b2f4-3c1e-7a8d-9f20-5d4c3b2a1e0f"
);

eventSource.onmessage = (event) => {
  const notification = JSON.parse(event.data);
  console.log(notification.title ?? "", notification.message);
};

eventSource.onerror = (error) => {
  console.error("Stream error:", error);
};

cURLCopy Anchor Link

curl -N https://app.alphorn.dev/api/stream/0199b2f4-3c1e-7a8d-9f20-5d4c3b2a1e0f

The -N flag disables buffering so events appear immediately. Press Ctrl+C to stop.

PythonCopy Anchor Link

import requests
import json

response = requests.get(
    "https://app.alphorn.dev/api/stream/0199b2f4-3c1e-7a8d-9f20-5d4c3b2a1e0f",
    stream=True,
    headers={"Accept": "text/event-stream"},
)

for line in response.iter_lines():
    if line:
        decoded = line.decode("utf-8")
        if decoded.startswith("data: "):
            notification = json.loads(decoded[6:])
            print(f"{notification.get('title', '')}: {notification['message']}")

GoCopy Anchor Link

package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
)

func main() {
	resp, err := http.Get("https://app.alphorn.dev/api/stream/0199b2f4-3c1e-7a8d-9f20-5d4c3b2a1e0f")
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	scanner := bufio.NewScanner(resp.Body)
	for scanner.Scan() {
		line := scanner.Text()
		if strings.HasPrefix(line, "data: ") {
			var notification map[string]interface{}
			json.Unmarshal([]byte(line[6:]), &notification)
			title, _ := notification["title"].(string) // missing when the notification has no title
			fmt.Printf("%s: %s\n", title, notification["message"])
		}
	}
}

Event FormatCopy Anchor Link

The channel's Output Format setting controls what each event looks like. The Include Priority, Include Tags, and Include Payload switches control whether those fields are included (all on by default).

JSONCopy Anchor Link

Each event is a JSON object in the data: line:

{
  "title": "Deploy Complete",
  "message": "v2.1.0 deployed to production",
  "timestamp": "2025-01-15T10:30:00.000Z",
  "priority": 3,
  "tags": ["deploy", "production"],
  "payload": { "title": "Deploy Complete", "message": "v2.1.0 deployed to production", "priority": 3, "tags": ["deploy", "production"] }
}

• message and timestamp are always present. timestamp is the time the event was streamed
• title is only present when the notification has one
• priority and tags are only present when switched on and when the notification has them
• payload is present when switched on. It is the full JSON request body, or {} for plain-text notifications

Plain textCopy Anchor Link

Each event is one line, with fields separated by |:

data: 2025-01-15T10:30:00.000Z | 3 | Deploy Complete | v2.1.0 deployed to production | deploy, production

The order is timestamp, priority, title, message, tags, payload. The connection examples above assume the JSON format - with text format, print event.data instead of parsing it. Fields that are switched off or empty are left out. In text format, the payload only contains the keys that are not already shown (everything except title, message, priority, and tags).

CommentsCopy Anchor Link

Right after connecting, the server sends a : connected comment, and then a : keepalive comment every 30 seconds. SSE clients ignore comments, but they keep proxies from closing idle connections.

Connection Limits and ErrorsCopy Anchor Link

Self-hosted instances can change the limits with SSE_MAX_CONNECTIONS_PER_CHANNEL and SSE_MAX_CONNECTIONS_PER_IP. The per-IP limit uses the first address in X-Forwarded-For (or X-Real-IP), so make sure your reverse proxy sets these headers correctly - without them, all clients count as the same IP and share 20 connections in total.

ReconnectionCopy Anchor Link

SSE connections may drop due to network issues, proxy timeouts, or server restarts. The browser's EventSource API reconnects automatically. For other clients, implement reconnection with backoff:

import time
import requests

def stream_with_reconnect(url, max_retries=10):
    retries = 0
    while retries < max_retries:
        try:
            with requests.get(
                url, stream=True, headers={"Accept": "text/event-stream"}, timeout=(10, 90)
            ) as response:
                if response.status_code == 404:
                    raise SystemExit("Stream not found - check the stream URL")
                response.raise_for_status()  # 429 / 503 -> retry with backoff below
                retries = 0  # reset after a successful connection
                for line in response.iter_lines():
                    if line:
                        print(line.decode("utf-8"))
        except requests.exceptions.RequestException as err:
            print(f"Stream error: {err}")
        retries += 1
        wait = min(30, 2 ** retries)
        print(f"Reconnecting in {wait}s...")
        time.sleep(wait)

The read timeout (90 seconds) is longer than the 30-second keepalive, so a silently dead connection is detected and reopened.

If you're running behind a reverse proxy, increase the read timeout to prevent the proxy from closing idle connections:

# Nginx
proxy_read_timeout 86400;
proxy_buffering off;

Self-HostingCopy Anchor Link

By default, streams are served by the Next.js web process itself (SSE_MODE unset). This works when the web app and the worker run in one process (MODE=all, the default).

Each process only streams to the clients connected to it. So use the standalone SSE server when:

• the worker runs separately (MODE=worker), or
• you run more than one web instance - otherwise a client only receives events that happen to be delivered by the instance it's connected to

There is exactly one SSE server (Alphorn publishes to the single SSE_SERVER_URL), so it can't be scaled out to several replicas. To set it up:

1. Set the following on the web app and the worker:

   Copy Text

   SSE_MODE=standalone
   SSE_SERVER_URL=http://sse:4000        # internal URL of the SSE server
   SSE_INTERNAL_SECRET=<random-secret>   # shared secret, e.g. openssl rand -hex 32

2. Start the SSE server from a source checkout (it is not part of the Docker image), with the same DATABASE_URL, SSE_INTERNAL_SECRET and SSE_PORT (default 4000):

   Copy Text

   pnpm install
   pnpm sse-server

   The server stops right away if SSE_INTERNAL_SECRET is missing. It accepts events up to 1 MB, so Live Stream deliveries of webhook bodies close to the 1 MiB ingest limit can fail in standalone mode.

3. In your reverse proxy, route the public path /api/stream/<id> to the SSE server's /stream/<id> (with buffering off, see Self-Hosting → Reverse Proxy). Do not expose the server's internal /publish/* endpoint publicly.

Use CasesCopy Anchor Link

• Live dashboards - Display notifications in real-time on a monitoring screen (served from the same domain, or through a proxy - see below)
• Custom notification UIs - Build your own notification center
• Automation - Trigger actions in response to notifications
• Logging - Stream notifications to a custom logging pipeline