Subscribe endpoint consolidation; same behavior for all endpoints; keepalive

README.md

@@ -64,13 +64,13 @@ sudo apt install ntfy
 **Debian/Ubuntu** (*manual install*)**:**
 ```bash
 sudo apt install tmux
-wget https://github.com/binwiederhier/ntfy/releases/download/v0.0.2/ntfy_0.0.2_amd64.deb
+wget https://github.com/binwiederhier/ntfy/releases/download/v0.0.4/ntfy_0.0.4_amd64.deb
-dpkg -i ntfy_0.0.2_amd64.deb
+dpkg -i ntfy_0.0.4_amd64.deb
 ```
 
 **Fedora/RHEL/CentOS:**
 ```bash
-rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v0.0.2/ntfy_0.0.2_amd64.rpm
+rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v0.0.4/ntfy_0.0.4_amd64.rpm
 ```
 
 **Docker:**
@@ -85,8 +85,8 @@ go get -u heckel.io/ntfy
 
 **Manual install** (*any x86_64-based Linux*)**:**
 ```bash
-wget https://github.com/binwiederhier/ntfy/releases/download/v0.0.2/ntfy_0.0.2_linux_x86_64.tar.gz
+wget https://github.com/binwiederhier/ntfy/releases/download/v0.0.4/ntfy_0.0.4_linux_x86_64.tar.gz
-sudo tar -C /usr/bin -zxf ntfy_0.0.2_linux_x86_64.tar.gz ntfy
+sudo tar -C /usr/bin -zxf ntfy_0.0.4_linux_x86_64.tar.gz ntfy
 ./ntfy
 ```
 
@@ -101,26 +101,10 @@ make build-simple
 To build releases, I use [GoReleaser](https://goreleaser.com/). If you have that installed, you can run `make build` or
 `make build-snapshot`.
 
-## FAQ
-
-### Isn't this like ...?
-Probably. I didn't do a whole lot of research before making this.
-
-### Can I use this in my app?
-Yes. As long as you don't abuse it, it'll be available and free of charge.
-
-### What are the uptime guarantees?
-Best effort.
-
-### Why is the web UI so ugly?
-I don't particularly like JS or dealing with CSS. I'll make it pretty after it's functional.
-
-### Will you know what topics exist, can you spy on me?
-If you don't trust me or your messages are sensitive, run your ntfy on your own server. That said, the logs do not 
-contain any topic names or other details about you.
-
 ## TODO
 - add HTTPS
+- make limits configurable
+- limit max number of subscriptions 
 
 ## Contributing
 I welcome any and all contributions. Just create a PR or an issue.

cmd/app.go

@@ -16,6 +16,7 @@ func New() *cli.App {
 	flags := []cli.Flag{
 		&cli.StringFlag{Name: "config", Aliases: []string{"c"}, EnvVars: []string{"NTFY_CONFIG_FILE"}, Value: "/etc/ntfy/config.yml", DefaultText: "/etc/ntfy/config.yml", Usage: "config file"},
 		altsrc.NewStringFlag(&cli.StringFlag{Name: "listen-http", Aliases: []string{"l"}, EnvVars: []string{"NTFY_LISTEN_HTTP"}, Value: config.DefaultListenHTTP, Usage: "ip:port used to as listen address"}),
+		altsrc.NewDurationFlag(&cli.DurationFlag{Name: "keepalive-interval", Aliases: []string{"k"}, EnvVars: []string{"NTFY_KEEPALIVE_INTERVAL"}, Value: config.DefaultKeepaliveInterval, Usage: "default interval of keepalive messages"}),
 	}
 	return &cli.App{
 		Name:                   "ntfy",
@@ -37,9 +38,11 @@ func New() *cli.App {
 func execRun(c *cli.Context) error {
 	// Read all the options
 	listenHTTP := c.String("listen-http")
+	keepaliveInterval := c.Duration("keepalive-interval")
 
 	// Run main bot, can be killed by signal
 	conf := config.New(listenHTTP)
+	conf.KeepaliveInterval = keepaliveInterval
 	s := server.New(conf)
 	if err := s.Run(); err != nil {
 		log.Fatalln(err)

config/config.go

@@ -8,8 +8,9 @@ import (
 
 // Defines default config settings
 const (
-	DefaultListenHTTP      = ":80"
+	DefaultListenHTTP        = ":80"
-	defaultManagerInterval = time.Minute
+	DefaultKeepaliveInterval = 30 * time.Second
+	defaultManagerInterval   = time.Minute
 )
 
 // Defines the max number of requests, here:
@@ -21,18 +22,20 @@ var (
 
 // Config is the main config struct for the application. Use New to instantiate a default config struct.
 type Config struct {
-	ListenHTTP      string
+	ListenHTTP        string
-	Limit           rate.Limit
+	Limit             rate.Limit
-	LimitBurst      int
+	LimitBurst        int
-	ManagerInterval time.Duration
+	KeepaliveInterval time.Duration
+	ManagerInterval   time.Duration
 }
 
 // New instantiates a default new config
 func New(listenHTTP string) *Config {
 	return &Config{
-		ListenHTTP:      listenHTTP,
+		ListenHTTP:        listenHTTP,
-		Limit:           defaultLimit,
+		Limit:             defaultLimit,
-		LimitBurst:      defaultLimitBurst,
+		LimitBurst:        defaultLimitBurst,
-		ManagerInterval: defaultManagerInterval,
+		KeepaliveInterval: DefaultKeepaliveInterval,
+		ManagerInterval:   defaultManagerInterval,
 	}
 }

examples/example_desktop_notifications.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+# This is an example shell script showing how to consume a ntfy.sh topic using
+# a simple script. The notify-send command sends any arriving message as a desktop notification.
+
+while read msg; do
+  notify-send "$msg"
+done < <(stdbuf -i0 -o0 curl -s ntfy.sh/mytopic/raw)

server/index.html

@@ -12,9 +12,9 @@
     <meta name="HandheldFriendly" content="true">
 
     <!-- Mobile browsers, background color -->
-    <meta name="theme-color" content="#004c79">
+    <meta name="theme-color" content="#39005a">
-    <meta name="msapplication-navbutton-color" content="#004c79">
+    <meta name="msapplication-navbutton-color" content="#39005a">
-    <meta name="apple-mobile-web-app-status-bar-style" content="#004c79">
+    <meta name="apple-mobile-web-app-status-bar-style" content="#39005a">
 
     <!-- Favicon, see favicon.io -->
     <link rel="icon" type="image/png" href="static/img/favicon.png">
@@ -25,14 +25,14 @@
     <meta property="og:site_name" content="ntfy.sh" />
     <meta property="og:title" content="ntfy.sh | simple HTTP-based pub-sub" />
     <meta property="og:description" content="ntfy is a simple HTTP-based pub-sub notification service. It allows you to send desktop notifications via scripts from any computer, entirely without signup or cost. Made with ❤ by Philipp C. Heckel, Apache License 2.0, source at https://heckel.io/ntfy." />
-    <meta property="og:image" content="/static/img/ntfy.png" />
+    <meta property="og:image" content="/static/img/favicon.png" />
     <meta property="og:url" content="https://ntfy.sh" />
 </head>
 <body>
 <div id="main">
     <h1>ntfy.sh - simple HTTP-based pub-sub</h1>
     <p>
-        <b>ntfy</b> (pronounce: <i>notify</i>) is a simple HTTP-based pub-sub notification service and tool.
+        <b>ntfy</b> (pronounce: <i>notify</i>) is a simple HTTP-based <a href="https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern">pub-sub</a> notification service.
         It allows you to send <b>desktop notifications via scripts from any computer</b>, entirely <b>without signup or cost</b>.
         It's also <a href="https://github.com/binwiederhier/ntfy">open source</a> if you want to run your own.
     </p>
@@ -55,33 +55,103 @@
     </p>
     <form id="subscribeForm">
         <p>
-            <label for="topicField">Topic ID:</label>
+            <label for="topicField">Subscribe to topic:</label>
-            <input type="text" id="topicField" placeholder="Letters, numbers, _ and -"  pattern="[-_A-Za-z]{1,64}" autofocus />
+            <input type="text" id="topicField" placeholder="Letters, numbers, _ and -"  pattern="[-_A-Za-z]{1,64}" />
             <input type="submit" id="subscribeButton" value="Subscribe" />
         </p>
     </form>
-    <p id="topicsHeader">Subscribed topics:</p>
+    <p id="topicsHeader">Topics:</p>
     <ul id="topicsList"></ul>
-    <audio id="notifySound" src="static/sound/mixkit-long-pop-2358.wav"></audio>
+    <audio id="notifySound" src="static/sound/mixkit-message-pop-alert-2354.mp3"></audio>
 
     <h3>Subscribe via your app, or via the CLI</h3>
+    <p class="smallMarginBottom">
+        Using <a href="https://developer.mozilla.org/en-US/docs/Web/API/EventSource">EventSource</a>, you can consume
+        notifications like this (see <a href="https://github.com/binwiederhier/ntfy/tree/main/examples">full example</a>):
+    </p>
     <code>
-        curl -s ntfy.sh/mytopic/raw # one message per line (\n are replaced with a space)<br/>
+        const eventSource = new EventSource('https://ntfy.sh/mytopic/sse');<br/>
-        curl -s ntfy.sh/mytopic/json # one JSON message per line<br/>
+        eventSource.onmessage = (e) => {<br/>
-        curl -s ntfy.sh/mytopic/sse # server-sent events (SSE) stream
+        &nbsp;&nbsp;// Do something with e.data<br/>
+        };
+    </code>
+    <p class="smallMarginBottom">
+        Or you can use <tt>curl</tt> or any other HTTP library. Here's an example for the <tt>/json</tt> endpoint,
+        which prints one JSON message per line (keepalive and open messages have an "event" field):
+    </p>
+    <code>
+        $ curl -s ntfy.sh/mytopic/json<br/>
+        {"time":1635359841,"event":"open"}<br/>
+        {"time":1635359844,"message":"This is a notification"}<br/>
+        {"time":1635359851,"event":"keepalive"}
+    </code>
+    <p class="smallMarginBottom">
+        Using the <tt>/sse</tt> endpoint (SSE, server-sent events stream):
+    </p>
+    <code>
+        $ curl -s ntfy.sh/mytopic/sse<br/>
+        event: open<br/>
+        data: {"time":1635359796,"event":"open"}<br/><br/>
+
+        data: {"time":1635359803,"message":"This is a notification"}<br/><br/>
+
+        event: keepalive<br/>
+        data: {"time":1635359806,"event":"keepalive"}
+    </code>
+    <p class="smallMarginBottom">
+        Using the <tt>/raw</tt> endpoint (empty lines are keepalive messages):
+    </p>
+    <code>
+        $ curl -s ntfy.sh/mytopic/raw<br/>
+        <br/>
+        This is a notification
     </code>
 
     <h2>Publishing messages</h2>
-    <p>
+    <p class="smallMarginBottom">
         Publishing messages can be done via PUT or POST using. Here's an example using <tt>curl</tt>:
     </p>
     <code>
         curl -d "long process is done" ntfy.sh/mytopic
     </code>
+    <p class="smallMarginBottom">
+        Here's an example in JS with <tt>fetch()</tt> (see <a href="https://github.com/binwiederhier/ntfy/tree/main/examples">full example</a>):
+    </p>
+    <code>
+        fetch('https://ntfy.sh/mytopic', {<br/>
+        &nbsp;&nbsp;method: 'POST', // PUT works too<br/>
+        &nbsp;&nbsp;body: 'Hello from the other side.'<br/>
+        })
+    </code>
     <p>
         Messages published to a non-existing topic or a topic without subscribers will not be delivered later.
         There is (currently) no buffering of any kind. If you're not listening, the message won't be delivered.
     </p>
+
+    <h2>FAQ</h2>
+    <p>
+        <b>Isn't this like ...?</b><br/>
+        Who knows. I didn't do a lot of research before making this. It was fun making it.
+    </p>
+
+    <p>
+        <b>Can I use this in my app? Will it stay free?</b><br/>
+        Yes. As long as you don't abuse it, it'll be available and free of charge. I do not plan on monetizing
+        the service.
+    </p>
+
+    <p>
+        <b>What are the uptime guarantees?</b><br/>
+        Best effort.
+    </p>
+
+    <p>
+        <b>Will you know what topics exist, can you spy on me?</b><br/>
+        If you don't trust me or your messages are sensitive, run your own server. It's <a href="https://github.com/binwiederhier/ntfy">open source</a>.
+        That said, the logs do not contain any topic names or other details about you. Check the code if you don't believe me.
+    </p>
+
+    <center id="ironicCenterTagDontFreakOut"><i>Made with ❤️ by <a href="https://heckel.io">Philipp C. Heckel</a></i></center>
 </div>
 <script src="static/js/app.js"></script>
 </body>

server/message.go

@@ -0,0 +1,43 @@
+package server
+
+import "time"
+
+// List of possible events
+const (
+	openEvent      = "open"
+	keepaliveEvent = "keepalive"
+)
+
+// message represents a message published to a topic
+type message struct {
+	Time    int64  `json:"time"`            // Unix time in seconds
+	Event   string `json:"event,omitempty"` // One of the above
+	Message string `json:"message,omitempty"`
+}
+
+// messageEncoder is a function that knows how to encode a message
+type messageEncoder func(msg *message) (string, error)
+
+// newMessage creates a new message with the current timestamp
+func newMessage(event string, msg string) *message {
+	return &message{
+		Time:    time.Now().Unix(),
+		Event:   event,
+		Message: msg,
+	}
+}
+
+// newOpenMessage is a convenience method to create an open message
+func newOpenMessage() *message {
+	return newMessage(openEvent, "")
+}
+
+// newKeepaliveMessage is a convenience method to create a keepalive message
+func newKeepaliveMessage() *message {
+	return newMessage(keepaliveEvent, "")
+}
+
+// newDefaultMessage is a convenience method to create a notification message
+func newDefaultMessage(msg string) *message {
+	return newMessage("", msg)
+}

server/server.go

@@ -48,11 +48,11 @@ const (
 )
 
 var (
-	topicRegex = regexp.MustCompile(`^/[^/]+$`)
+	topicRegex  = regexp.MustCompile(`^/[^/]+$`)
-	jsonRegex  = regexp.MustCompile(`^/[^/]+/json$`)
+	jsonRegex   = regexp.MustCompile(`^/[^/]+/json$`)
-	sseRegex   = regexp.MustCompile(`^/[^/]+/sse$`)
+	sseRegex    = regexp.MustCompile(`^/[^/]+/sse$`)
-	rawRegex   = regexp.MustCompile(`^/[^/]+/raw$`)
+	rawRegex    = regexp.MustCompile(`^/[^/]+/raw$`)
-	staticRegex   = regexp.MustCompile(`^/static/.+`)
+	staticRegex = regexp.MustCompile(`^/static/.+`)
 
 	//go:embed "index.html"
 	indexSource string
@@ -159,11 +159,7 @@ func (s *Server) handlePublishHTTP(w http.ResponseWriter, r *http.Request) error
 	if err != nil {
 		return err
 	}
-	msg := &message{
+	if err := t.Publish(newDefaultMessage(string(b))); err != nil {
-		Time:    time.Now().UnixMilli(),
-		Message: string(b),
-	}
-	if err := t.Publish(msg); err != nil {
 		return err
 	}
 	w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
@@ -171,75 +167,74 @@ func (s *Server) handlePublishHTTP(w http.ResponseWriter, r *http.Request) error
 }
 
 func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request) error {
-	t := s.createTopic(strings.TrimSuffix(r.URL.Path[1:], "/json")) // Hack
+	encoder := func(msg *message) (string, error) {
-	subscriberID := t.Subscribe(func(msg *message) error {
+		var buf bytes.Buffer
-		if err := json.NewEncoder(w).Encode(&msg); err != nil {
+		if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
-			return err
+			return "", err
 		}
-		if fl, ok := w.(http.Flusher); ok {
+		return buf.String(), nil
-			fl.Flush()
-		}
-		return nil
-	})
-	defer s.unsubscribe(t, subscriberID)
-	w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
-	select {
-	case <-t.ctx.Done():
-	case <-r.Context().Done():
 	}
-	return nil
+	return s.handleSubscribe(w, r, "json", "application/stream+json", encoder)
 }
 
 func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request) error {
-	t := s.createTopic(strings.TrimSuffix(r.URL.Path[1:], "/sse")) // Hack
+	encoder := func(msg *message) (string, error) {
-	subscriberID := t.Subscribe(func(msg *message) error {
 		var buf bytes.Buffer
 		if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
-			return err
+			return "", err
 		}
-		m := fmt.Sprintf("data: %s\n", buf.String())
+		if msg.Event != "" {
-		if _, err := io.WriteString(w, m); err != nil {
+			return fmt.Sprintf("event: %s\ndata: %s\n", msg.Event, buf.String()), nil // Browser's .onmessage() does not fire on this!
-			return err
 		}
-		if fl, ok := w.(http.Flusher); ok {
+		return fmt.Sprintf("data: %s\n", buf.String()), nil
-			fl.Flush()
-		}
-		return nil
-	})
-	defer s.unsubscribe(t, subscriberID)
-	w.Header().Set("Content-Type", "text/event-stream")
-	w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
-	if _, err := io.WriteString(w, "event: open\n\n"); err != nil {
-		return err
 	}
-	if fl, ok := w.(http.Flusher); ok {
+	return s.handleSubscribe(w, r, "sse", "text/event-stream", encoder)
-		fl.Flush()
-	}
-	select {
-	case <-t.ctx.Done():
-	case <-r.Context().Done():
-	}
-	return nil
 }
 
 func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request) error {
-	t := s.createTopic(strings.TrimSuffix(r.URL.Path[1:], "/raw")) // Hack
+	encoder := func(msg *message) (string, error) {
-	subscriberID := t.Subscribe(func(msg *message) error {
+		if msg.Event == "" { // only handle default events
-		m := strings.ReplaceAll(msg.Message, "\n", " ") + "\n"
+			return strings.ReplaceAll(msg.Message, "\n", " ") + "\n", nil
-		if _, err := io.WriteString(w, m); err != nil {
+		}
+		return "\n", nil // "keepalive" and "open" events just send an empty line
+	}
+	return s.handleSubscribe(w, r, "raw", "text/plain", encoder)
+}
+
+func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, format string, contentType string, encoder messageEncoder) error {
+	t := s.createTopic(strings.TrimSuffix(r.URL.Path[1:], "/"+format)) // Hack
+	sub := func(msg *message) error {
+		m, err := encoder(msg)
+		if err != nil {
+			return err
+		}
+		if _, err := w.Write([]byte(m)); err != nil {
 			return err
 		}
 		if fl, ok := w.(http.Flusher); ok {
 			fl.Flush()
 		}
 		return nil
-	})
-	defer s.unsubscribe(t, subscriberID)
-	select {
-	case <-t.ctx.Done():
-	case <-r.Context().Done():
 	}
-	return nil
+	subscriberID := t.Subscribe(sub)
+	defer s.unsubscribe(t, subscriberID)
+	w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
+	w.Header().Set("Content-Type", contentType)
+	if err := sub(newOpenMessage()); err != nil { // Send out open message
+		return err
+	}
+	for {
+		select {
+		case <-t.ctx.Done():
+			return nil
+		case <-r.Context().Done():
+			return nil
+		case <-time.After(s.config.KeepaliveInterval):
+			if err := sub(newKeepaliveMessage()); err != nil { // Send keepalive message
+				return err
+			}
+		}
+	}
 }
 
 func (s *Server) handleOptions(w http.ResponseWriter, r *http.Request) error {

server/static/css/app.css

@@ -20,6 +20,7 @@ h1 {
     font-size: 2.5em;
 }
 
+
 h2 {
     margin-top: 20px;
     margin-bottom: 5px;
@@ -33,8 +34,14 @@ h3 {
 }
 
 p {
-    margin-top: 0;
+    margin-top: 10px;
+    margin-bottom: 20px;
     font-size: 1.1em;
+    line-height: 140%;
+}
+
+p.smallMarginBottom {
+    margin-bottom: 10px;
 }
 
 tt {
@@ -49,6 +56,8 @@ code {
     font-family: monospace;
     padding: 20px;
     border-radius: 3px;
+    margin-top: 10px;
+    margin-bottom: 20px;
 }
 
 /* Lato font (OFL), https://fonts.google.com/specimen/Lato#about,
@@ -74,3 +83,7 @@ code {
     color: darkred;
     font-style: italic;
 }
+
+#ironicCenterTagDontFreakOut {
+    color: #666;
+}

server/static/js/app.js

@@ -59,8 +59,11 @@ const subscribeInternal = (topic, delaySec) => {
         };
         eventSource.onmessage = (e) => {
             const event = JSON.parse(e.data);
-            new Notification(event.message);
             notifySound.play();
+            new Notification(`${location.host}/${topic}`, {
+                body: event.message,
+                icon: '/static/img/favicon.png'
+            });
         };
         topics[topic] = eventSource;
         localStorage.setItem('topics', JSON.stringify(Object.keys(topics)));
@@ -80,7 +83,7 @@ const unsubscribe = (topic) => {
 const test = (topic) => {
     fetch(`/${topic}`, {
         method: 'PUT',
-        body: `This is a test notification for topic ${topic}!`
+        body: `This is a test notification`
     })
 };
 

server/topic.go

@@ -21,15 +21,10 @@ type topic struct {
 	mu          sync.Mutex
 }
 
-// message represents a message published to a topic
-type message struct {
-	Time    int64  `json:"time"`
-	Message string `json:"message"`
-}
-
 // subscriber is a function that is called for every new message on a topic
 type subscriber func(msg *message) error
 
+// newTopic creates a new topic
 func newTopic(id string) *topic {
 	ctx, cancel := context.WithCancel(context.Background())
 	return &topic{