Bump

.goreleaser.yml

@@ -4,11 +4,9 @@ before:
 builds:
   - binary: ntfy
     env:
-      - CGO_ENABLED=0
+      - CGO_ENABLED=1 # required for go-sqlite3
     goos:
       - linux
-      - windows
-      - darwin
     goarch:
       - amd64
 nfpms:

Makefile

@@ -94,6 +94,7 @@ build-snapshot:
 
 build-simple: clean
 	mkdir -p dist/ntfy_linux_amd64
+	export CGO_ENABLED=1
 	$(GO) build \
 		-o dist/ntfy_linux_amd64/ntfy \
 		-ldflags \

README.md

@@ -1,47 +1,112 @@
-# ntfy
+![ntfy](server/static/img/ntfy.png)
 
-ntfy (pronounce: *notify*) is a super simple pub-sub notification service. It allows you to send desktop and (soon) phone notifications
-via scripts. I run a free version of it on *[ntfy.sh](https://ntfy.sh)*. **No signups or cost.**
+# ntfy - simple HTTP-based pub-sub
+
+**ntfy** (pronounce: *notify*) is a simple HTTP-based [pub-sub](https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern) notification service.
+It allows you to **send notifications to your phone or desktop via scripts** from any computer, entirely **without signup or cost**.
+It's also open source (as you can plainly see) if you want to run your own.
+
+I run a free version of it at **[ntfy.sh](https://ntfy.sh)**, and there's an [Android app](https://play.google.com/store/apps/details?id=io.heckel.ntfy)
+too.
 
 ## Usage
 
-### Subscribe to a topic
-Topics are created on the fly by subscribing to them. You can create and subscribe to a topic either in a web UI, or in 
-your own app by subscribing to an [SSE](https://en.wikipedia.org/wiki/Server-sent_events)/[EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource),
-or a JSON or raw feed.  
+### Publishing messages
 
-Because there is no sign-up, **the topic is essentially a password**, so pick something that's not easily guessable.  
+Publishing messages can be done via PUT or POST using. Topics are created on the fly by subscribing or publishing to them.
+Because there is no sign-up, **the topic is essentially a password**, so pick something that's not easily guessable.
 
-Here's how you can create a topic `mytopic`, subscribe to it topic and wait for events. This is using `curl`, but you
-can use any library that can do HTTP GETs:
+Here's an example showing how to publish a message using `curl`:
 
-```
-# Subscribe to "mytopic" and output one message per line (\n are replaced with a space)
-curl -s ntfy.sh/mytopic/raw
-
-# Subscribe to "mytopic" and output one JSON message per line
-curl -s ntfy.sh/mytopic/json
-
-# Subscribe to "mytopic" and output an SSE stream (supported via JS/EventSource)
-curl -s ntfy.sh/mytopic/sse
-```
-
-You can easily script it to execute any command when a message arrives. This sends desktop notifications (just like 
-the web UI, but without it):
-```
-while read msg; do
-  [ -n "$msg" ] && notify-send "$msg"
-done < <(stdbuf -i0 -o0 curl -s ntfy.sh/mytopic/raw)
-```
-
-### Publish messages
-Publishing messages can be done via PUT or POST using. Here's an example using `curl`:
 ```
 curl -d "long process is done" ntfy.sh/mytopic
 ```
 
-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.
+Here's an example in JS with `fetch()` (see [full example](examples)):
+
+```
+fetch('https://ntfy.sh/mytopic', {
+  method: 'POST', // PUT works too
+  body: 'Hello from the other side.'
+})
+```
+
+### Subscribe to a topic
+You can create and subscribe to a topic either in this web UI, or in your own app by subscribing to an
+[EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource), a JSON feed, or raw feed.
+
+#### Subscribe via web
+If you subscribe to a topic via this web UI in the field below, messages published to any subscribed topic
+will show up as **desktop notification**.
+
+You can try this easily on **[ntfy.sh](https://ntfy.sh)**.
+
+#### Subscribe via phone
+You can use the [Ntfy Android App](https://play.google.com/store/apps/details?id=io.heckel.ntfy) to receive 
+notifications directly on your phone. Just like the server, this app is also [open source](https://github.com/binwiederhier/ntfy-android).
+
+#### Subscribe via your app, or via the CLI
+Using [EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource) in JS, you can consume
+notifications like this (see [full example](examples)):
+
+```javascript
+const eventSource = new EventSource('https://ntfy.sh/mytopic/sse');<br/>
+eventSource.onmessage = (e) => {<br/>
+  // Do something with e.data<br/>
+};
+```
+
+You can also use the same `/sse` endpoint via `curl` or any other HTTP library:
+
+```
+$ curl -s ntfy.sh/mytopic/sse
+event: open
+data: {"id":"weSj9RtNkj","time":1635528898,"event":"open","topic":"mytopic"}
+
+data: {"id":"p0M5y6gcCY","time":1635528909,"event":"message","topic":"mytopic","message":"Hi!"}
+
+event: keepalive
+data: {"id":"VNxNIg5fpt","time":1635528928,"event":"keepalive","topic":"test"}
+```
+
+To consume JSON instead, use the `/json` endpoint, which prints one message per line:
+
+```
+$ curl -s ntfy.sh/mytopic/json
+{"id":"SLiKI64DOt","time":1635528757,"event":"open","topic":"mytopic"}
+{"id":"hwQ2YpKdmg","time":1635528741,"event":"message","topic":"mytopic","message":"Hi!"}
+{"id":"DGUDShMCsc","time":1635528787,"event":"keepalive","topic":"mytopic"}
+```
+
+Or use the `/raw` endpoint if you need something super simple (empty lines are keepalive messages):
+
+```
+$ curl -s ntfy.sh/mytopic/raw
+
+This is a notification
+```
+
+#### Message buffering and polling
+Messages are buffered in memory for a few hours to account for network interruptions of subscribers.
+You can read back what you missed by using the `since=...` query parameter. It takes either a
+duration (e.g. `10m` or `30s`) or a Unix timestamp (e.g. `1635528757`):
+
+```
+$ curl -s "ntfy.sh/mytopic/json?since=10m"
+# Same output as above, but includes messages from up to 10 minutes ago
+```
+
+You can also just poll for messages if you don't like the long-standing connection using the `poll=1`
+query parameter. The connection will end after all available messages have been read. This parameter has to be
+combined with `since=`.
+
+```
+$ curl -s "ntfy.sh/mytopic/json?poll=1&since=10m"
+# Returns messages from up to 10 minutes ago and ends the connection
+```
+
+## Examples
+There are a few usage examples in the [examples](examples) directory. I'm sure there are tons of other ways to use it.
 
 ## Installation
 Please check out the [releases page](https://github.com/binwiederhier/ntfy/releases) for binaries and
@@ -64,13 +129,13 @@ sudo apt install ntfy
 **Debian/Ubuntu** (*manual install*)**:**
 ```bash
 sudo apt install tmux
-wget https://github.com/binwiederhier/ntfy/releases/download/v1.1.2/ntfy_1.1.2_amd64.deb
-dpkg -i ntfy_1.1.2_amd64.deb
+wget https://github.com/binwiederhier/ntfy/releases/download/v1.2.4/ntfy_1.2.4_amd64.deb
+dpkg -i ntfy_1.2.4_amd64.deb
 ```
 
 **Fedora/RHEL/CentOS:**
 ```bash
-rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v1.1.2/ntfy_1.1.2_amd64.rpm
+rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v1.2.4/ntfy_1.2.4_amd64.rpm
 ```
 
 **Docker:**
@@ -85,8 +150,8 @@ go get -u heckel.io/ntfy
 
 **Manual install** (*any x86_64-based Linux*)**:**
 ```bash
-wget https://github.com/binwiederhier/ntfy/releases/download/v1.1.2/ntfy_1.1.2_linux_x86_64.tar.gz
-sudo tar -C /usr/bin -zxf ntfy_1.1.2_linux_x86_64.tar.gz ntfy
+wget https://github.com/binwiederhier/ntfy/releases/download/v1.2.4/ntfy_1.2.4_linux_x86_64.tar.gz
+sudo tar -C /usr/bin -zxf ntfy_1.2.4_linux_x86_64.tar.gz ntfy
 ./ntfy
 ```
 
@@ -104,7 +169,6 @@ To build releases, I use [GoReleaser](https://goreleaser.com/). If you have that
 ## 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.
@@ -116,4 +180,6 @@ Third party libraries and resources:
 * [github.com/urfave/cli/v2](https://github.com/urfave/cli/v2) (MIT) is used to drive the CLI
 * [Mixkit sound](https://mixkit.co/free-sound-effects/notification/) (Mixkit Free License) used as notification sound
 * [Lato Font](https://www.latofonts.com/) (OFL) is used as a font in the Web UI
-* [GoReleaser](https://goreleaser.com/) (MIT) is used to create releases 
+* [GoReleaser](https://goreleaser.com/) (MIT) is used to create releases
+* [github.com/mattn/go-sqlite3](https://github.com/mattn/go-sqlite3) (MIT) is used to provide the persistent message cache
+* [Firebase Admin SDK](https://github.com/firebase/firebase-admin-go) (Apache 2.0) is used to send FCM messages

cmd/app.go

@@ -20,9 +20,15 @@ func New() *cli.App {
 		&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.NewStringFlag(&cli.StringFlag{Name: "firebase-key-file", Aliases: []string{"F"}, EnvVars: []string{"NTFY_FIREBASE_KEY_FILE"}, Usage: "Firebase credentials file; if set additionally publish to FCM topic"}),
-		altsrc.NewDurationFlag(&cli.DurationFlag{Name: "message-buffer-duration", Aliases: []string{"b"}, EnvVars: []string{"NTFY_MESSAGE_BUFFER_DURATION"}, Value: config.DefaultMessageBufferDuration, Usage: "buffer messages in memory for this time to allow `since` requests"}),
-		altsrc.NewDurationFlag(&cli.DurationFlag{Name: "keepalive-interval", Aliases: []string{"k"}, EnvVars: []string{"NTFY_KEEPALIVE_INTERVAL"}, Value: config.DefaultKeepaliveInterval, Usage: "default interval of keepalive messages"}),
-		altsrc.NewDurationFlag(&cli.DurationFlag{Name: "manager-interval", Aliases: []string{"m"}, EnvVars: []string{"NTFY_MANAGER_INTERVAL"}, Value: config.DefaultManagerInterval, Usage: "default interval of for message pruning and stats printing"}),
+		altsrc.NewStringFlag(&cli.StringFlag{Name: "cache-file", Aliases: []string{"C"}, EnvVars: []string{"NTFY_CACHE_FILE"}, Usage: "cache file used for message caching"}),
+		altsrc.NewDurationFlag(&cli.DurationFlag{Name: "cache-duration", Aliases: []string{"b"}, EnvVars: []string{"NTFY_CACHE_DURATION"}, Value: config.DefaultCacheDuration, Usage: "buffer messages for this time to allow `since` requests"}),
+		altsrc.NewDurationFlag(&cli.DurationFlag{Name: "keepalive-interval", Aliases: []string{"k"}, EnvVars: []string{"NTFY_KEEPALIVE_INTERVAL"}, Value: config.DefaultKeepaliveInterval, Usage: "interval of keepalive messages"}),
+		altsrc.NewDurationFlag(&cli.DurationFlag{Name: "manager-interval", Aliases: []string{"m"}, EnvVars: []string{"NTFY_MANAGER_INTERVAL"}, Value: config.DefaultManagerInterval, Usage: "interval of for message pruning and stats printing"}),
+		altsrc.NewIntFlag(&cli.IntFlag{Name: "global-topic-limit", Aliases: []string{"T"}, EnvVars: []string{"NTFY_GLOBAL_TOPIC_LIMIT"}, Value: config.DefaultGlobalTopicLimit, Usage: "total number of topics allowed"}),
+		altsrc.NewIntFlag(&cli.IntFlag{Name: "visitor-subscription-limit", Aliases: []string{"V"}, EnvVars: []string{"NTFY_VISITOR_SUBSCRIPTION_LIMIT"}, Value: config.DefaultVisitorSubscriptionLimit, Usage: "number of subscriptions per visitor"}),
+		altsrc.NewIntFlag(&cli.IntFlag{Name: "visitor-request-limit-burst", Aliases: []string{"B"}, EnvVars: []string{"NTFY_VISITOR_REQUEST_LIMIT_BURST"}, Value: config.DefaultVisitorRequestLimitBurst, Usage: "initial limit of requests per visitor"}),
+		altsrc.NewDurationFlag(&cli.DurationFlag{Name: "visitor-request-limit-replenish", Aliases: []string{"R"}, EnvVars: []string{"NTFY_VISITOR_REQUEST_LIMIT_REPLENISH"}, Value: config.DefaultVisitorRequestLimitReplenish, Usage: "interval at which burst limit is replenished (one per x)"}),
+		altsrc.NewBoolFlag(&cli.BoolFlag{Name: "behind-proxy", Aliases: []string{"P"}, EnvVars: []string{"NTFY_BEHIND_PROXY"}, Value: false, Usage: "if set, use X-Forwarded-For header to determine visitor IP address (for rate limiting)"}),
 	}
 	return &cli.App{
 		Name:                   "ntfy",
@@ -45,9 +51,15 @@ func execRun(c *cli.Context) error {
 	// Read all the options
 	listenHTTP := c.String("listen-http")
 	firebaseKeyFile := c.String("firebase-key-file")
-	messageBufferDuration := c.Duration("message-buffer-duration")
+	cacheFile := c.String("cache-file")
+	cacheDuration := c.Duration("cache-duration")
 	keepaliveInterval := c.Duration("keepalive-interval")
 	managerInterval := c.Duration("manager-interval")
+	globalTopicLimit := c.Int("global-topic-limit")
+	visitorSubscriptionLimit := c.Int("visitor-subscription-limit")
+	visitorRequestLimitBurst := c.Int("visitor-request-limit-burst")
+	visitorRequestLimitReplenish := c.Duration("visitor-request-limit-replenish")
+	behindProxy := c.Bool("behind-proxy")
 
 	// Check values
 	if firebaseKeyFile != "" && !util.FileExists(firebaseKeyFile) {
@@ -56,16 +68,22 @@ func execRun(c *cli.Context) error {
 		return errors.New("keepalive interval cannot be lower than five seconds")
 	} else if managerInterval < 5*time.Second {
 		return errors.New("manager interval cannot be lower than five seconds")
-	} else if messageBufferDuration < managerInterval {
-		return errors.New("message buffer duration cannot be lower than manager interval")
+	} else if cacheDuration < managerInterval {
+		return errors.New("cache duration cannot be lower than manager interval")
 	}
 
 	// Run server
 	conf := config.New(listenHTTP)
 	conf.FirebaseKeyFile = firebaseKeyFile
-	conf.MessageBufferDuration = messageBufferDuration
+	conf.CacheFile = cacheFile
+	conf.CacheDuration = cacheDuration
 	conf.KeepaliveInterval = keepaliveInterval
 	conf.ManagerInterval = managerInterval
+	conf.GlobalTopicLimit = globalTopicLimit
+	conf.VisitorSubscriptionLimit = visitorSubscriptionLimit
+	conf.VisitorRequestLimitBurst = visitorRequestLimitBurst
+	conf.VisitorRequestLimitReplenish = visitorRequestLimitReplenish
+	conf.BehindProxy = behindProxy
 	s, err := server.New(conf)
 	if err != nil {
 		log.Fatalln(err)

config/config.go

@@ -2,53 +2,56 @@
 package config
 
 import (
-	"golang.org/x/time/rate"
 	"time"
 )
 
 // Defines default config settings
 const (
-	DefaultListenHTTP            = ":80"
-	DefaultMessageBufferDuration = 12 * time.Hour
-	DefaultKeepaliveInterval     = 30 * time.Second
-	DefaultManagerInterval       = time.Minute
+	DefaultListenHTTP        = ":80"
+	DefaultCacheDuration     = 12 * time.Hour
+	DefaultKeepaliveInterval = 30 * time.Second
+	DefaultManagerInterval   = time.Minute
 )
 
 // Defines all the limits
-// - request limit: max number of PUT/GET/.. requests (here: 50 requests bucket, replenished at a rate of 1 per second)
 // - global topic limit: max number of topics overall
-// - subscription limit: max number of subscriptions (active HTTP connections) per per-visitor/IP
-var (
-	defaultGlobalTopicLimit         = 5000
-	defaultVisitorRequestLimit      = rate.Every(time.Second)
-	defaultVisitorRequestLimitBurst = 50
-	defaultVisitorSubscriptionLimit = 30
+// - per visistor request limit: max number of PUT/GET/.. requests (here: 60 requests bucket, replenished at a rate of one per 10 seconds)
+// - per visistor subscription limit: max number of subscriptions (active HTTP connections) per per-visitor/IP
+const (
+	DefaultGlobalTopicLimit             = 5000
+	DefaultVisitorRequestLimitBurst     = 60
+	DefaultVisitorRequestLimitReplenish = 10 * time.Second
+	DefaultVisitorSubscriptionLimit     = 30
 )
 
 // Config is the main config struct for the application. Use New to instantiate a default config struct.
 type Config struct {
-	ListenHTTP               string
-	FirebaseKeyFile          string
-	MessageBufferDuration    time.Duration
-	KeepaliveInterval        time.Duration
-	ManagerInterval          time.Duration
-	GlobalTopicLimit         int
-	VisitorRequestLimit      rate.Limit
-	VisitorRequestLimitBurst int
-	VisitorSubscriptionLimit int
+	ListenHTTP                   string
+	FirebaseKeyFile              string
+	CacheFile                    string
+	CacheDuration                time.Duration
+	KeepaliveInterval            time.Duration
+	ManagerInterval              time.Duration
+	GlobalTopicLimit             int
+	VisitorRequestLimitBurst     int
+	VisitorRequestLimitReplenish time.Duration
+	VisitorSubscriptionLimit     int
+	BehindProxy                  bool
 }
 
 // New instantiates a default new config
 func New(listenHTTP string) *Config {
 	return &Config{
-		ListenHTTP:               listenHTTP,
-		FirebaseKeyFile:          "",
-		MessageBufferDuration:    DefaultMessageBufferDuration,
-		KeepaliveInterval:        DefaultKeepaliveInterval,
-		ManagerInterval:          DefaultManagerInterval,
-		GlobalTopicLimit:         defaultGlobalTopicLimit,
-		VisitorRequestLimit:      defaultVisitorRequestLimit,
-		VisitorRequestLimitBurst: defaultVisitorRequestLimitBurst,
-		VisitorSubscriptionLimit: defaultVisitorSubscriptionLimit,
+		ListenHTTP:                   listenHTTP,
+		FirebaseKeyFile:              "",
+		CacheFile:                    "",
+		CacheDuration:                DefaultCacheDuration,
+		KeepaliveInterval:            DefaultKeepaliveInterval,
+		ManagerInterval:              DefaultManagerInterval,
+		GlobalTopicLimit:             DefaultGlobalTopicLimit,
+		VisitorRequestLimitBurst:     DefaultVisitorRequestLimitBurst,
+		VisitorRequestLimitReplenish: DefaultVisitorRequestLimitReplenish,
+		VisitorSubscriptionLimit:     DefaultVisitorSubscriptionLimit,
+		BehindProxy:                  false,
 	}
 }

config/config.yml

@@ -10,16 +10,45 @@
 #
 # firebase-key-file: <filename>
 
+# If set, messages are cached in a local SQLite database instead of only in-memory. This
+# allows for service restarts without losing messages in support of the since= parameter.
+#
+# cache-file: <filename>
+
 # Duration for which messages will be buffered before they are deleted.
 # This is required to support the "since=..." and "poll=1" parameter.
 #
-# message-buffer-duration: 12h
+# cache-duration: 12h
 
 # Interval in which keepalive messages are sent to the client. This is to prevent
 # intermediaries closing the connection for inactivity.
 #
 # keepalive-interval: 30s
 
-# Interval in which the manager prunes old messages, deletes topics and prints the stats.
+# Interval in which the manager prunes old messages, deletes topics
+# and prints the stats.
 #
 # manager-interval: 1m
+
+# Rate limiting: Total number of topics before the server rejects new topics.
+#
+# global-topic-limit: 5000
+
+# Rate limiting: Number of subscriptions per visitor (IP address)
+#
+# visitor-subscription-limit: 30
+
+# Rate limiting: Allowed GET/PUT/POST requests per second, per visitor:
+# - visitor-request-limit-burst is the initial bucket of requests each visitor has
+# - visitor-request-limit-replenish is the rate at which the bucket is refilled
+#
+# visitor-request-limit-burst: 60
+# visitor-request-limit-replenish: 10s
+
+# If set, the X-Forwarded-For header is used to determine the visitor IP address
+# instead of the remote address of the connection.
+#
+# WARNING: If you are behind a proxy, you must set this, otherwise all visitors are rate limited
+#          as if they are one.
+#
+# behind-proxy: false

examples/example_desktop_notifications.sh

@@ -3,5 +3,5 @@
 # a simple script. The notify-send command sends any arriving message as a desktop notification.
 
 while read msg; do
-  notify-send "$msg"
+  [ -n "$msg" ] && notify-send "$msg"
 done < <(stdbuf -i0 -o0 curl -s ntfy.sh/mytopic/raw)

go.mod

@@ -1,16 +1,42 @@
 module heckel.io/ntfy
 
-go 1.16
+go 1.17
 
 require (
-	cloud.google.com/go/firestore v1.6.0 // indirect
+	cloud.google.com/go/firestore v1.6.1 // indirect
 	cloud.google.com/go/storage v1.18.2 // indirect
 	firebase.google.com/go v3.13.0+incompatible
 	github.com/BurntSushi/toml v0.4.1 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.1 // indirect
+	github.com/mattn/go-sqlite3 v1.14.9
 	github.com/urfave/cli/v2 v2.3.0
 	golang.org/x/oauth2 v0.0.0-20211028175245-ba495a64dcb5 // indirect
 	golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac
 	google.golang.org/api v0.60.0
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )
+
+require (
+	cloud.google.com/go v0.97.0 // indirect
+	github.com/census-instrumentation/opencensus-proto v0.3.0 // indirect
+	github.com/cespare/xxhash v1.1.0 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4 // indirect
+	github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1 // indirect
+	github.com/envoyproxy/go-control-plane v0.10.0 // indirect
+	github.com/envoyproxy/protoc-gen-validate v0.6.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/google/go-cmp v0.5.6 // indirect
+	github.com/googleapis/gax-go/v2 v2.1.1 // indirect
+	github.com/russross/blackfriday/v2 v2.1.0 // indirect
+	go.opencensus.io v0.23.0 // indirect
+	golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d // indirect
+	golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359 // indirect
+	golang.org/x/text v0.3.7 // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
+	google.golang.org/appengine v1.6.7 // indirect
+	google.golang.org/genproto v0.0.0-20211101144312-62acf1d99145 // indirect
+	google.golang.org/grpc v1.41.0 // indirect
+	google.golang.org/protobuf v1.27.1 // indirect
+)

go.sum

@@ -36,6 +36,8 @@ cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7
 cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1b3c64qFpCk=
 cloud.google.com/go/firestore v1.6.0 h1:dMIWvm+3O0E3DM7kcZPH0FBQ94Xg/OMkdTNDaY9itbI=
 cloud.google.com/go/firestore v1.6.0/go.mod h1:afJwI0vaXwAG54kI7A//lP/lSPDkQORQuMkv56TxEPU=
+cloud.google.com/go/firestore v1.6.1 h1:8rBq3zRjnHx8UtBvaOWqBB1xq9jH6/wltfQLlTMh2Fw=
+cloud.google.com/go/firestore v1.6.1/go.mod h1:asNXNOzBdyVQmEU+ggO8UPodTkEVFW5Qx+rwHnAz+EY=
 cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
 cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
 cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
@@ -59,8 +61,13 @@ github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAE
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/census-instrumentation/opencensus-proto v0.2.1 h1:glEXhBS5PSLLv4IXzLA5yPRVX4bilULVyxxbrfOtDAk=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
+github.com/census-instrumentation/opencensus-proto v0.3.0 h1:t/LhUZLVitR1Ow2YOnduCsavhwFUklBMoGVYUCqmCqk=
+github.com/census-instrumentation/opencensus-proto v0.3.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
+github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
+github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
@@ -69,8 +76,15 @@ github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGX
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403 h1:cqQfy1jclcSy/FwLjemeg3SR1yaINm74aQyupQ0Bl8M=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
+github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4 h1:hzAQntlaYRkVSFEfj9OTWlVV1H155FMD8BTKktLv0QI=
+github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
 github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed h1:OZmjad4L3H8ncOIR8rnb5MREYqG8ixi5+WbeUsquF0c=
 github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/cncf/xds/go v0.0.0-20211001041855-01bcc9b48dfe/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1 h1:zH8ljVhhq7yC0MIeUL/IviMtY8hx2mK8cN9wEYb8ggw=
+github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/cpuguy83/go-md2man/v2 v2.0.1 h1:r/myEWzV9lfsM1tFLgDyu0atFtJ1fXn261LKYj/3DxU=
 github.com/cpuguy83/go-md2man/v2 v2.0.1/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
@@ -83,8 +97,13 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0 h1:dulLQAYQFYtG5MTplgNGHWuV2D+OBD+Z8lmDBmbLg+s=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
+github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0=
+github.com/envoyproxy/go-control-plane v0.10.0 h1:WVt4HEPbdRbRD/PKKPbPnIVavO6gk/h673jWyIJ016k=
+github.com/envoyproxy/go-control-plane v0.10.0/go.mod h1:AY7fTTXNdv/aJ2O5jwpxAPOWUZ7hQAEvzN5Pf27BkQQ=
 github.com/envoyproxy/protoc-gen-validate v0.1.0 h1:EQciDnbrYxy13PgWoY8AqoxGiPrpgBZ1R8UNe3ddc+A=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/envoyproxy/protoc-gen-validate v0.6.2 h1:JiO+kJTpmYGjEodY7O1Zk8oZcNz1+f30UtwtXoFUPzE=
+github.com/envoyproxy/protoc-gen-validate v0.6.2/go.mod h1:2t7qjJNvHPx8IjnBOzl9E9/baC+qXE/TeeyBRzgJDws=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
@@ -94,6 +113,8 @@ github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4er
 github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e h1:1r7pUrabqp18hOBcwBwiTsbnFeTZHV9eER/QT5JVZxY=
 github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
+github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE=
+github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y=
@@ -168,16 +189,24 @@ github.com/googleapis/gax-go/v2 v2.1.1/go.mod h1:hddJymUZASv3XPyGkUpKj8pPO47Rmb0
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/iancoleman/strcase v0.2.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47ZCWhYzw7ho=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
 github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/lyft/protoc-gen-star v0.5.3/go.mod h1:V0xaHgaf5oCCqmcxYcWiDfTiKsZsRc87/1qhoTACD8w=
+github.com/mattn/go-sqlite3 v1.14.8/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
+github.com/mattn/go-sqlite3 v1.14.9 h1:10HX2Td0ocZpYEjhilsuo6WWtUqttj2Kb0KtD86/KYA=
+github.com/mattn/go-sqlite3 v1.14.9/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZI=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
@@ -188,10 +217,13 @@ github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
+github.com/spf13/afero v1.3.3/go.mod h1:5KUK8ByomD5Ti5Artl0RtHeI5pTF7MIDuXL3yY520V4=
+github.com/spf13/afero v1.6.0/go.mod h1:Ai8FlHk4v/PARR026UzYexafAt9roJ7LcLMAmO6Z93I=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/urfave/cli/v2 v2.3.0 h1:qph92Y649prgesehzOrQjdWyxFOp/QVM+6imKHad91M=
 github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI=
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@@ -211,6 +243,7 @@ go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqe
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -248,6 +281,7 @@ golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.5.0/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -284,6 +318,8 @@ golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLd
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420 h1:a8jGStKg0XqKDlKqjLrXn0ioF5MH36pT7Z0BRTqLhbk=
 golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d h1:LO7XpTYMwTqxjLcGWPijK3vRXg1aWdlNOVOHRq45d7c=
+golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -356,9 +392,11 @@ golang.org/x/sys v0.0.0-20210603125802-9665404d3644/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210806184541-e5e7981a1069/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210917161153-d61c044b1678/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359 h1:2B5p2L5IfGiD7+b9BOoRMC6DgObAVZV+Fsp050NqXik=
 golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -371,6 +409,8 @@ golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -462,6 +502,7 @@ google.golang.org/api v0.55.0/go.mod h1:38yMfeP1kfjsl8isn0tliTjIb1rJXcQi4UXlbqiv
 google.golang.org/api v0.56.0/go.mod h1:38yMfeP1kfjsl8isn0tliTjIb1rJXcQi4UXlbqivdVE=
 google.golang.org/api v0.57.0/go.mod h1:dVPlbZyBo2/OjBpmvNdpn2GRm6rPy75jyU7bmhdrMgI=
 google.golang.org/api v0.58.0/go.mod h1:cAbP2FsxoGVNwtgNAmmn3y5G1TWAiVYRmg4yku3lv+E=
+google.golang.org/api v0.59.0/go.mod h1:sT2boj7M9YJxZzgeZqXogmhfmRWDtPzT31xkieUbuZU=
 google.golang.org/api v0.60.0 h1:eq/zs5WPH4J9undYM9IP1O7dSr7Yh8Y0GtSCpzGzIUk=
 google.golang.org/api v0.60.0/go.mod h1:d7rl65NZAkEQ90JFzqBjcRq1TVeG5ZoGV3sSpEnnVb4=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
@@ -529,9 +570,13 @@ google.golang.org/genproto v0.0.0-20210903162649-d08c68adba83/go.mod h1:eFjDcFEc
 google.golang.org/genproto v0.0.0-20210909211513-a8c4777a87af/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
 google.golang.org/genproto v0.0.0-20210917145530-b395a37504d4/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
 google.golang.org/genproto v0.0.0-20210924002016-3dee208752a0/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
+google.golang.org/genproto v0.0.0-20211008145708-270636b82663/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
 google.golang.org/genproto v0.0.0-20211016002631-37fc39342514/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
 google.golang.org/genproto v0.0.0-20211021150943-2b146023228c h1:FqrtZMB5Wr+/RecOM3uPJNPfWR8Upb5hAPnt7PU6i4k=
 google.golang.org/genproto v0.0.0-20211021150943-2b146023228c/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
+google.golang.org/genproto v0.0.0-20211028162531-8db9c33dc351/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
+google.golang.org/genproto v0.0.0-20211101144312-62acf1d99145 h1:vum3nDKdleYb+aePXKFEDT2+ghuH00EgYp9B7Q7EZZE=
+google.golang.org/genproto v0.0.0-20211101144312-62acf1d99145/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
@@ -558,6 +603,8 @@ google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnD
 google.golang.org/grpc v1.39.1/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE=
 google.golang.org/grpc v1.40.0 h1:AGJ0Ih4mHjSeibYkFGh1dD9KJ/eOtZ93I6hoHhukQ5Q=
 google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
+google.golang.org/grpc v1.41.0 h1:f+PlOh7QV4iIJkPrx5NQ7qaNGFQ3OTse67yaDHfju4E=
+google.golang.org/grpc v1.41.0/go.mod h1:U3l9uK9J0sini8mHphKoXyaqDA/8VyGnDee1zzIUK6k=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0/go.mod h1:6Kw0yEErY5E/yWrBtf03jp27GLLJujG4z/JK95pnjjw=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=

server/cache.go

@@ -0,0 +1,14 @@
+package server
+
+import (
+	_ "github.com/mattn/go-sqlite3" // SQLite driver
+	"time"
+)
+
+type cache interface {
+	AddMessage(m *message) error
+	Messages(topic string, since time.Time) ([]*message, error)
+	MessageCount(topic string) (int, error)
+	Topics() (map[string]*topic, error)
+	Prune(keep time.Duration) error
+}

server/cache_mem.go

@@ -0,0 +1,80 @@
+package server
+
+import (
+	_ "github.com/mattn/go-sqlite3" // SQLite driver
+	"sync"
+	"time"
+)
+
+type memCache struct {
+	messages map[string][]*message
+	mu       sync.Mutex
+}
+
+var _ cache = (*memCache)(nil)
+
+func newMemCache() *memCache {
+	return &memCache{
+		messages: make(map[string][]*message),
+	}
+}
+
+func (s *memCache) AddMessage(m *message) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if _, ok := s.messages[m.Topic]; !ok {
+		s.messages[m.Topic] = make([]*message, 0)
+	}
+	s.messages[m.Topic] = append(s.messages[m.Topic], m)
+	return nil
+}
+
+func (s *memCache) Messages(topic string, since time.Time) ([]*message, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if _, ok := s.messages[topic]; !ok {
+		return make([]*message, 0), nil
+	}
+	messages := make([]*message, 0) // copy!
+	for _, m := range s.messages[topic] {
+		msgTime := time.Unix(m.Time, 0)
+		if msgTime == since || msgTime.After(since) {
+			messages = append(messages, m)
+		}
+	}
+	return messages, nil
+}
+
+func (s *memCache) MessageCount(topic string) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if _, ok := s.messages[topic]; !ok {
+		return 0, nil
+	}
+	return len(s.messages[topic]), nil
+}
+
+func (s *memCache) Topics() (map[string]*topic, error) {
+	// Hack since we know when this is called there are no messages!
+	return make(map[string]*topic), nil
+}
+
+func (s *memCache) Prune(keep time.Duration) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for topic, _ := range s.messages {
+		s.pruneTopic(topic, keep)
+	}
+	return nil
+}
+
+func (s *memCache) pruneTopic(topic string, keep time.Duration) {
+	for i, m := range s.messages[topic] {
+		msgTime := time.Unix(m.Time, 0)
+		if time.Since(msgTime) < keep {
+			s.messages[topic] = s.messages[topic][i:]
+			return
+		}
+	}
+	s.messages[topic] = make([]*message, 0) // all messages expired
+}

server/cache_sqlite.go

@@ -0,0 +1,127 @@
+package server
+
+import (
+	"database/sql"
+	"errors"
+	_ "github.com/mattn/go-sqlite3" // SQLite driver
+	"time"
+)
+
+const (
+	createTableQuery = `
+		BEGIN;
+		CREATE TABLE IF NOT EXISTS messages (
+			id VARCHAR(20) PRIMARY KEY,
+			time INT NOT NULL,
+			topic VARCHAR(64) NOT NULL,
+			message VARCHAR(1024) NOT NULL
+		);
+		CREATE INDEX IF NOT EXISTS idx_topic ON messages (topic);
+		COMMIT;
+	`
+	insertMessageQuery           = `INSERT INTO messages (id, time, topic, message) VALUES (?, ?, ?, ?)`
+	pruneMessagesQuery           = `DELETE FROM messages WHERE time < ?`
+	selectMessagesSinceTimeQuery = `
+		SELECT id, time, message 
+		FROM messages 
+		WHERE topic = ? AND time >= ?
+		ORDER BY time ASC
+	`
+	selectMessageCountQuery = `SELECT count(*) FROM messages WHERE topic = ?`
+	selectTopicsQuery       = `SELECT topic, MAX(time) FROM messages GROUP BY TOPIC`
+)
+
+type sqliteCache struct {
+	db *sql.DB
+}
+
+var _ cache = (*sqliteCache)(nil)
+
+func newSqliteCache(filename string) (*sqliteCache, error) {
+	db, err := sql.Open("sqlite3", filename)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := db.Exec(createTableQuery); err != nil {
+		return nil, err
+	}
+	return &sqliteCache{
+		db: db,
+	}, nil
+}
+
+func (c *sqliteCache) AddMessage(m *message) error {
+	_, err := c.db.Exec(insertMessageQuery, m.ID, m.Time, m.Topic, m.Message)
+	return err
+}
+
+func (c *sqliteCache) Messages(topic string, since time.Time) ([]*message, error) {
+	rows, err := c.db.Query(selectMessagesSinceTimeQuery, topic, since.Unix())
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	messages := make([]*message, 0)
+	for rows.Next() {
+		var timestamp int64
+		var id, msg string
+		if err := rows.Scan(&id, &timestamp, &msg); err != nil {
+			return nil, err
+		}
+		messages = append(messages, &message{
+			ID:      id,
+			Time:    timestamp,
+			Event:   messageEvent,
+			Topic:   topic,
+			Message: msg,
+		})
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return messages, nil
+}
+
+func (c *sqliteCache) MessageCount(topic string) (int, error) {
+	rows, err := c.db.Query(selectMessageCountQuery, topic)
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	var count int
+	if !rows.Next() {
+		return 0, errors.New("no rows found")
+	}
+	if err := rows.Scan(&count); err != nil {
+		return 0, err
+	} else if err := rows.Err(); err != nil {
+		return 0, err
+	}
+	return count, nil
+}
+
+func (s *sqliteCache) Topics() (map[string]*topic, error) {
+	rows, err := s.db.Query(selectTopicsQuery)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	topics := make(map[string]*topic, 0)
+	for rows.Next() {
+		var id string
+		var last int64
+		if err := rows.Scan(&id, &last); err != nil {
+			return nil, err
+		}
+		topics[id] = newTopic(id, time.Unix(last, 0))
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return topics, nil
+}
+
+func (c *sqliteCache) Prune(keep time.Duration) error {
+	_, err := c.db.Exec(pruneMessagesQuery, time.Now().Add(-1*keep).Unix())
+	return err
+}

server/index.html

@@ -33,10 +33,14 @@
     <h1><img src="static/img/ntfy.png" alt="ntfy"/><br/>ntfy.sh - simple HTTP-based pub-sub</h1>
     <p>
         <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.
+        It allows you to send notifications <a href="https://play.google.com/store/apps/details?id=io.heckel.ntfy">to your phone</a> or desktop via scripts from any computer,
+        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>
+    <p>
+        There are many ways to use ntfy. You can send yourself messages for all sorts of things: When a long process finishes or fails (a backup, a long rsync job, ...),
+        or to notify yourself when somebody logs into your server(s). Or you may want to use it in your own app to distribute messages to subscribed clients.
+        Endless possibilities 😀.
     </p>
-    <p id="error"></p>
 
     <h2>Publishing messages</h2>
     <p>
@@ -65,26 +69,29 @@
         <a href="https://developer.mozilla.org/en-US/docs/Web/API/EventSource">EventSource</a>, a JSON feed, or raw feed.
     </p>
 
-    <h3>Subscribe via web</h3>
-    <p>
-        If you subscribe to a topic via this web UI in the field below, messages published to any subscribed topic
-        will show up as <b>desktop notification</b>.
-    </p>
-    <form id="subscribeForm">
+    <div id="subscribeBox">
+        <h3>Subscribe in this Web UI</h3>
+        <p id="error"></p>
         <p>
-            <label for="topicField">Subscribe to topic:</label>
-            <input type="text" id="topicField" placeholder="Letters, numbers, _ and -"  pattern="[-_A-Za-z]{1,64}" />
-            <input type="submit" id="subscribeButton" value="Subscribe" />
+            Subscribe to topics here and receive messages as <b>desktop notification</b>. Topics are not password-protected,
+            so choose a name that's not easy to guess. Once subscribed, you can publish messages via PUT/POST.
         </p>
-    </form>
-    <p id="topicsHeader">Topics:</p>
-    <ul id="topicsList"></ul>
-    <audio id="notifySound" src="static/sound/mixkit-message-pop-alert-2354.mp3"></audio>
+        <form id="subscribeForm">
+            <p>
+                <b>Topic:</b><br/>
+                <input type="text" id="topicField" autocomplete="off" placeholder="Topic name, e.g. phil_alerts"  pattern="[-_A-Za-z]{1,64}" />
+                <button id="subscribeButton">Subscribe</button>
+            </p>
+            <p id="topicsHeader"><b>Subscribed topics:</b></p>
+            <ul id="topicsList"></ul>
+        </form>
+        <audio id="notifySound" src="static/sound/mixkit-message-pop-alert-2354.mp3"></audio>
+    </div>
 
-    <h3>Subscribe via phone</h3>
+    <h3>Subscribe via Android App</h3>
     <p>
-        Once it's approved, you can use the <b>Ntfy Android App</b> to receive notifications directly on your phone. Just like
-        the server, this app is also <a href="https://github.com/binwiederhier/ntfy-android">open source</a>.
+        You can use the <a href="https://play.google.com/store/apps/details?id=io.heckel.ntfy">Ntfy Android App</a>
+        to receive notifications directly on your phone. Just like the server, this app is also <a href="https://github.com/binwiederhier/ntfy-android">open source</a>.
     </p>
 
     <h3>Subscribe via your app, or via the CLI</h3>
@@ -128,6 +135,14 @@
         <br/>
         This is a notification
     </code>
+    <p class="smallMarginBottom">
+        Here's an example of how to use this endpoint to send desktop notifications for every incoming message:
+    </p>
+    <code>
+        while read msg; do<br/>
+        &nbsp;&nbsp;[ -n "$msg" ] && notify-send "$msg"<br/>
+        done < <(stdbuf -i0 -o0 curl -s ntfy.sh/mytopic/raw)
+    </code>
 
     <h3>Message buffering and polling</h3>
     <p class="smallMarginBottom">
@@ -177,21 +192,19 @@
         In addition to caching messages locally and delivering them to long-polling subscribers, all messages are also
         published to Firebase Cloud Messaging (FCM) (if <tt>FirebaseKeyFile</tt> is set, which it is on ntfy.sh). This
         is to facilitate instant notifications on Android. I tried really, really hard to avoid using FCM, but newer
-        versions of Android made it impossible to implement <a href="https://developer.android.com/guide/background">background services</a>>.
-        I'm sorry.
+        versions of Android made it impossible to implement <a href="https://developer.android.com/guide/background">background services</a>.
     </p>
 
     <h2>Privacy policy</h2>
     <p>
         Neither the server nor the app record any personal information, or share any of the messages and topics with
-        any outside service. All data is exclusively used to make the service function properly. The notable exception
+        any outside service. All data is exclusively used to make the service function properly. The one exception
         is the Firebase Cloud Messaging (FCM) service, which is required to provide instant Android notifications (see
         FAQ for details).
     </p>
-
     <p>
         The web server does not log or otherwise store request paths, remote IP addresses or even topics or messages,
-        aside from a short on-disk cache (up to a day) to support the <tt>since=</tt> feature and service restarts.
+        aside from a short on-disk cache (up to a day) to support service restarts.
     </p>
 
     <center id="ironicCenterTagDontFreakOut"><i>Made with ❤️ by <a href="https://heckel.io">Philipp C. Heckel</a></i></center>

server/server.go

@@ -23,7 +23,6 @@ import (
 )
 
 // TODO add "max messages in a topic" limit
-// TODO implement persistence
 // TODO implement "since=<ID>"
 
 // Server is the main server
@@ -33,6 +32,7 @@ type Server struct {
 	visitors map[string]*visitor
 	firebase subscriber
 	messages int64
+	cache    cache
 	mu       sync.Mutex
 }
 
@@ -47,7 +47,7 @@ func (e errHTTP) Error() string {
 }
 
 const (
-	messageLimit = 1024
+	messageLimit = 512
 )
 
 var (
@@ -78,14 +78,35 @@ func New(conf *config.Config) (*Server, error) {
 			return nil, err
 		}
 	}
+	cache, err := createCache(conf)
+	if err != nil {
+		return nil, err
+	}
+	topics, err := cache.Topics()
+	if err != nil {
+		return nil, err
+	}
+	for _, t := range topics {
+		if firebaseSubscriber != nil {
+			t.Subscribe(firebaseSubscriber)
+		}
+	}
 	return &Server{
 		config:   conf,
+		cache:    cache,
 		firebase: firebaseSubscriber,
-		topics:   make(map[string]*topic),
+		topics:   topics,
 		visitors: make(map[string]*visitor),
 	}, nil
 }
 
+func createCache(conf *config.Config) (cache, error) {
+	if conf.CacheFile != "" {
+		return newSqliteCache(conf.CacheFile)
+	}
+	return newMemCache(), nil
+}
+
 func createFirebaseSubscriber(conf *config.Config) (subscriber, error) {
 	fb, err := firebase.NewApp(context.Background(), nil, option.WithCredentialsFile(conf.FirebaseKeyFile))
 	if err != nil {
@@ -138,24 +159,22 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request) error {
-	v := s.visitor(r.RemoteAddr)
-	if err := v.RequestAllowed(); err != nil {
-		return err
-	}
 	if r.Method == http.MethodGet && r.URL.Path == "/" {
 		return s.handleHome(w, r)
+	} else if r.Method == http.MethodHead && r.URL.Path == "/" {
+		return s.handleEmpty(w, r)
 	} else if r.Method == http.MethodGet && staticRegex.MatchString(r.URL.Path) {
 		return s.handleStatic(w, r)
-	} else if (r.Method == http.MethodPut || r.Method == http.MethodPost) && topicRegex.MatchString(r.URL.Path) {
-		return s.handlePublish(w, r, v)
-	} else if r.Method == http.MethodGet && jsonRegex.MatchString(r.URL.Path) {
-		return s.handleSubscribeJSON(w, r, v)
-	} else if r.Method == http.MethodGet && sseRegex.MatchString(r.URL.Path) {
-		return s.handleSubscribeSSE(w, r, v)
-	} else if r.Method == http.MethodGet && rawRegex.MatchString(r.URL.Path) {
-		return s.handleSubscribeRaw(w, r, v)
 	} else if r.Method == http.MethodOptions {
 		return s.handleOptions(w, r)
+	} else if (r.Method == http.MethodPut || r.Method == http.MethodPost) && topicRegex.MatchString(r.URL.Path) {
+		return s.withRateLimit(w, r, s.handlePublish)
+	} else if r.Method == http.MethodGet && jsonRegex.MatchString(r.URL.Path) {
+		return s.withRateLimit(w, r, s.handleSubscribeJSON)
+	} else if r.Method == http.MethodGet && sseRegex.MatchString(r.URL.Path) {
+		return s.withRateLimit(w, r, s.handleSubscribeSSE)
+	} else if r.Method == http.MethodGet && rawRegex.MatchString(r.URL.Path) {
+		return s.withRateLimit(w, r, s.handleSubscribeRaw)
 	}
 	return errHTTPNotFound
 }
@@ -165,6 +184,10 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request) error {
 	return err
 }
 
+func (s *Server) handleEmpty(w http.ResponseWriter, r *http.Request) error {
+	return nil
+}
+
 func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request) error {
 	http.FileServer(http.FS(webStaticFs)).ServeHTTP(w, r)
 	return nil
@@ -180,10 +203,17 @@ func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request, v *visito
 	if err != nil {
 		return err
 	}
-	if err := t.Publish(newDefaultMessage(t.id, string(b))); err != nil {
+	m := newDefaultMessage(t.id, string(b))
+	if err := t.Publish(m); err != nil {
+		return err
+	}
+	if err := s.cache.AddMessage(m); err != nil {
 		return err
 	}
 	w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
+	if err := json.NewEncoder(w).Encode(m); err != nil {
+		return err
+	}
 	s.mu.Lock()
 	s.messages++
 	s.mu.Unlock()
@@ -255,20 +285,18 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, v *visi
 	w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
 	w.Header().Set("Content-Type", contentType)
 	if poll {
-		return sendOldMessages(t, since, sub)
+		return s.sendOldMessages(t, since, sub)
 	}
 	subscriberID := t.Subscribe(sub)
 	defer t.Unsubscribe(subscriberID)
 	if err := sub(newOpenMessage(t.id)); err != nil { // Send out open message
 		return err
 	}
-	if err := sendOldMessages(t, since, sub); err != nil {
+	if err := s.sendOldMessages(t, since, sub); err != nil {
 		return err
 	}
 	for {
 		select {
-		case <-t.ctx.Done():
-			return nil
 		case <-r.Context().Done():
 			return nil
 		case <-time.After(s.config.KeepaliveInterval):
@@ -280,11 +308,15 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, v *visi
 	}
 }
 
-func sendOldMessages(t *topic, since time.Time, sub subscriber) error {
+func (s *Server) sendOldMessages(t *topic, since time.Time, sub subscriber) error {
 	if since.IsZero() {
 		return nil
 	}
-	for _, m := range t.Messages(since) {
+	messages, err := s.cache.Messages(t.id, since)
+	if err != nil {
+		return err
+	}
+	for _, m := range messages {
 		if err := sub(m); err != nil {
 			return err
 		}
@@ -318,7 +350,7 @@ func (s *Server) topic(id string) (*topic, error) {
 		if len(s.topics) >= s.config.GlobalTopicLimit {
 			return nil, errHTTPTooManyRequests
 		}
-		s.topics[id] = newTopic(id)
+		s.topics[id] = newTopic(id, time.Now())
 		if s.firebase != nil {
 			s.topics[id].Subscribe(s.firebase)
 		}
@@ -337,35 +369,54 @@ func (s *Server) updateStatsAndExpire() {
 		}
 	}
 
-	// Prune old messages, remove subscriptions without subscribers
-	for _, t := range s.topics {
-		t.Prune(s.config.MessageBufferDuration)
-		subs, msgs := t.Stats()
-		if msgs == 0 && (subs == 0 || (s.firebase != nil && subs == 1)) { // Firebase is a subscriber!
-			delete(s.topics, t.id)
-		}
+	// Prune cache
+	if err := s.cache.Prune(s.config.CacheDuration); err != nil {
+		log.Printf("error pruning cache: %s", err.Error())
 	}
 
-	// Print stats
+	// Prune old messages, remove subscriptions without subscribers
 	var subscribers, messages int
 	for _, t := range s.topics {
-		subs, msgs := t.Stats()
+		subs := t.Subscribers()
+		msgs, err := s.cache.MessageCount(t.id)
+		if err != nil {
+			log.Printf("cannot get stats for topic %s: %s", t.id, err.Error())
+			continue
+		}
+		if msgs == 0 && (subs == 0 || (s.firebase != nil && subs == 1)) { // Firebase is a subscriber!
+			delete(s.topics, t.id)
+			continue
+		}
 		subscribers += subs
 		messages += msgs
 	}
+
+	// Print stats
 	log.Printf("Stats: %d message(s) published, %d topic(s) active, %d subscriber(s), %d message(s) buffered, %d visitor(s)",
 		s.messages, len(s.topics), subscribers, messages, len(s.visitors))
 }
 
+func (s *Server) withRateLimit(w http.ResponseWriter, r *http.Request, handler func(w http.ResponseWriter, r *http.Request, v *visitor) error) error {
+	v := s.visitor(r)
+	if err := v.RequestAllowed(); err != nil {
+		return err
+	}
+	return handler(w, r, v)
+}
+
 // visitor creates or retrieves a rate.Limiter for the given visitor.
 // This function was taken from https://www.alexedwards.net/blog/how-to-rate-limit-http-requests (MIT).
-func (s *Server) visitor(remoteAddr string) *visitor {
+func (s *Server) visitor(r *http.Request) *visitor {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	remoteAddr := r.RemoteAddr
 	ip, _, err := net.SplitHostPort(remoteAddr)
 	if err != nil {
 		ip = remoteAddr // This should not happen in real life; only in tests.
 	}
+	if s.config.BehindProxy && r.Header.Get("X-Forwarded-For") != "" {
+		ip = r.Header.Get("X-Forwarded-For")
+	}
 	v, exists := s.visitors[ip]
 	if !exists {
 		s.visitors[ip] = newVisitor(s.config)

server/static/css/app.css

@@ -58,6 +58,7 @@ code {
     border-radius: 3px;
     margin-top: 10px;
     margin-bottom: 20px;
+    overflow-x: auto;
 }
 
 /* Lato font (OFL), https://fonts.google.com/specimen/Lato#about,
@@ -87,3 +88,159 @@ code {
 #ironicCenterTagDontFreakOut {
     color: #666;
 }
+
+/* Subscribe box */
+
+button {
+    background: #3a9784;
+    border: none;
+    border-radius: 3px;
+    padding: 3px 5px;
+    color: white;
+    cursor: pointer;
+}
+
+button:hover {
+    background: #317f6f;
+    padding: 5px;
+}
+
+ul {
+    padding-left: 1em;
+    list-style-type: none;
+    padding-bottom: 0;
+    margin: 0;
+}
+
+li {
+    padding: 4px 0;
+    margin: 4px 0;
+    font-size: 0.9em;
+}
+
+/* Subscribe box SMALL SCREEN */
+@media only screen and (max-width: 1599px) {
+    #subscribeBox #subscribeForm {
+        border-left: 4px solid #3a9784;
+        padding: 10px;
+    }
+
+    #subscribeBox #topicsHeader {
+        margin-bottom: 0;
+    }
+
+    #subscribeBox input {
+        height: 24px;
+        min-width: 200px;
+        max-width: 300px;
+        border-radius: 3px;
+        border: none;
+        border-bottom: 1px solid #aaa;
+        font-size: 0.8em;
+    }
+
+    #subscribeBox input:focus {
+        border-bottom: 2px solid #3a9784;
+        outline: none;
+    }
+
+    #subscribeBox ul {
+        margin: 0;
+        padding: 0;
+    }
+
+    #subscribeBox li {
+        margin: 3px 0;
+        padding: 0;
+    }
+
+    #subscribeBox li img {
+        width: 15px;
+        height: 15px;
+        vertical-align: bottom;
+    }
+
+    #subscribeBox button {
+        font-size: 0.8em;
+        background: #3a9784;
+        border-radius: 3px;
+        padding: 5px;
+        color: white;
+        cursor: pointer;
+    }
+
+    #subscribeBox button:hover {
+        background: #317f6f;
+    }
+}
+
+/* Subscribe box BIG SCREEN */
+@media only screen and (min-width: 1600px) {
+    #subscribeBox {
+        position: fixed;
+        top: 170px;
+        right: 10px;
+        width: 300px;
+        border-left: 4px solid #3a9784;
+        padding: 10px;
+    }
+
+    #subscribeBox h3 {
+        margin-top: 0;
+        margin-bottom: 5px;
+        font-size: 1.1em;
+    }
+
+    #subscribeBox #topicsHeader {
+        margin-bottom: 0;
+    }
+
+    #subscribeBox p {
+        font-size: 0.9em;
+        margin-bottom: 10px;
+    }
+
+    #subscribeBox ul {
+        margin: 0;
+        padding: 0;
+    }
+
+    #subscribeBox input {
+        height: 18px;
+        border-radius: 3px;
+        border: none;
+        border-bottom: 1px solid #aaa;
+    }
+
+    #subscribeBox input:focus {
+        border-bottom: 2px solid #3a9784;
+        outline: none;
+    }
+
+    #subscribeBox li {
+        margin: 3px 0;
+        padding: 0;
+    }
+
+    #subscribeBox li img {
+        width: 15px;
+        height: 15px;
+        vertical-align: bottom;
+    }
+
+    #subscribeBox button {
+        font-size: 0.7em;
+        background: #3a9784;
+        border-radius: 3px;
+        padding: 5px;
+        color: white;
+        cursor: pointer;
+    }
+
+    #subscribeBox button:hover {
+        background: #317f6f;
+    }
+
+}
+
+

server/static/img/clear_black_24dp.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" height="24px" viewBox="0 0 24 24" width="24px" fill="#FFFFFF"><path d="M0 0h24v24H0V0z" fill="none"/><path d="M19 6.41L17.59 5 12 10.59 6.41 5 5 6.41 10.59 12 5 17.59 6.41 19 12 13.41 17.59 19 19 17.59 13.41 12 19 6.41z"/></svg>

server/static/img/send_black_24dp.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" height="24px" viewBox="0 0 24 24" width="24px" fill="#ffffff"><path d="M0 0h24v24H0z" fill="none"/><path d="M2.01 21L23 12 2.01 3 2 10l15 2-15 2z"/></svg>

server/static/js/app.js

@@ -16,7 +16,6 @@ const topicsList = document.getElementById("topicsList");
 const topicField = document.getElementById("topicField");
 const notifySound = document.getElementById("notifySound");
 const subscribeButton = document.getElementById("subscribeButton");
-const subscribeForm = document.getElementById("subscribeForm");
 const errorField = document.getElementById("error");
 
 const subscribe = (topic) => {
@@ -40,7 +39,7 @@ const subscribeInternal = (topic, delaySec) => {
         if (!topicEntry) {
             topicEntry = document.createElement('li');
             topicEntry.id = `topic-${topic}`;
-            topicEntry.innerHTML = `${topic} <button onclick="test('${topic}')">Test</button> <button onclick="unsubscribe('${topic}')">Unsubscribe</button>`;
+            topicEntry.innerHTML = `${topic} <button onclick="test('${topic}'); return false;"> <img src="static/img/send_black_24dp.svg"> Test</button> <button onclick="unsubscribe('${topic}'); return false;"> <img src="static/img/clear_black_24dp.svg"> Unsubscribe</button>`;
             topicsList.appendChild(topicEntry);
         }
         topicsHeader.style.display = '';
@@ -48,13 +47,13 @@ const subscribeInternal = (topic, delaySec) => {
         // Open event source
         let eventSource = new EventSource(`${topic}/sse`);
         eventSource.onopen = () => {
-            topicEntry.innerHTML = `${topic} <button onclick="test('${topic}')">Test</button> <button onclick="unsubscribe('${topic}')">Unsubscribe</button>`;
+            topicEntry.innerHTML = `${topic} <button onclick="test('${topic}'); return false;"> <img src="static/img/send_black_24dp.svg"> Test</button> <button onclick="unsubscribe('${topic}'); return false;"> <img src="static/img/clear_black_24dp.svg"> Unsubscribe</button>`;
             delaySec = 0; // Reset on successful connection
         };
         eventSource.onerror = (e) => {
+            topicEntry.innerHTML = `${topic} <i>(Reconnecting)</i> <button disabled="disabled">Test</button> <button onclick="unsubscribe('${topic}'); return false;">Unsubscribe</button>`;
+            eventSource.close();
             const newDelaySec = (delaySec + 5 <= 15) ? delaySec + 5 : 15;
-            topicEntry.innerHTML = `${topic} <i>(Reconnecting in ${newDelaySec}s ...)</i> <button disabled="disabled">Test</button> <button onclick="unsubscribe('${topic}')">Unsubscribe</button>`;
-            eventSource.close()
             subscribeInternal(topic, newDelaySec);
         };
         eventSource.onmessage = (e) => {
@@ -83,7 +82,7 @@ const unsubscribe = (topic) => {
 const test = (topic) => {
     fetch(`/${topic}`, {
         method: 'PUT',
-        body: `This is a test notification`
+        body: `This is a test notification sent by the ntfy.sh Web UI at ${new Date().toString()}.`
     })
 };
 
@@ -101,7 +100,7 @@ const showNotificationDeniedError = () => {
     showError("You have blocked desktop notifications for this website. Please unblock them and refresh to use the web-based desktop notifications.");
 };
 
-subscribeForm.onsubmit = function () {
+subscribeButton.onclick = function () {
     if (!topicField.value) {
         return false;
     }

server/topic.go

@@ -1,7 +1,6 @@
 package server
 
 import (
-	"context"
 	"log"
 	"math/rand"
 	"sync"
@@ -12,11 +11,8 @@ import (
 // can publish a message
 type topic struct {
 	id          string
-	subscribers map[int]subscriber
-	messages    []*message
 	last        time.Time
-	ctx         context.Context
-	cancel      context.CancelFunc
+	subscribers map[int]subscriber
 	mu          sync.Mutex
 }
 
@@ -24,14 +20,11 @@ type topic struct {
 type subscriber func(msg *message) error
 
 // newTopic creates a new topic
-func newTopic(id string) *topic {
-	ctx, cancel := context.WithCancel(context.Background())
+func newTopic(id string, last time.Time) *topic {
 	return &topic{
 		id:          id,
+		last:        last,
 		subscribers: make(map[int]subscriber),
-		last:        time.Now(),
-		ctx:         ctx,
-		cancel:      cancel,
 	}
 }
 
@@ -54,7 +47,6 @@ func (t *topic) Publish(m *message) error {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	t.last = time.Now()
-	t.messages = append(t.messages, m)
 	for _, s := range t.subscribers {
 		if err := s(m); err != nil {
 			log.Printf("error publishing message to subscriber")
@@ -63,38 +55,8 @@ func (t *topic) Publish(m *message) error {
 	return nil
 }
 
-func (t *topic) Messages(since time.Time) []*message {
+func (t *topic) Subscribers() int {
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	messages := make([]*message, 0) // copy!
-	for _, m := range t.messages {
-		msgTime := time.Unix(m.Time, 0)
-		if msgTime == since || msgTime.After(since) {
-			messages = append(messages, m)
-		}
-	}
-	return messages
-}
-
-func (t *topic) Prune(keep time.Duration) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	for i, m := range t.messages {
-		msgTime := time.Unix(m.Time, 0)
-		if time.Since(msgTime) < keep {
-			t.messages = t.messages[i:]
-			return
-		}
-	}
-	t.messages = make([]*message, 0)
-}
-
-func (t *topic) Stats() (subscribers int, messages int) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	return len(t.subscribers), len(t.messages)
-}
-
-func (t *topic) Close() {
-	t.cancel()
+	return len(t.subscribers)
 }

server/visitor.go

@@ -24,7 +24,7 @@ type visitor struct {
 func newVisitor(conf *config.Config) *visitor {
 	return &visitor{
 		config:        conf,
-		limiter:       rate.NewLimiter(conf.VisitorRequestLimit, conf.VisitorRequestLimitBurst),
+		limiter:       rate.NewLimiter(rate.Every(conf.VisitorRequestLimitReplenish), conf.VisitorRequestLimitBurst),
 		subscriptions: util.NewLimiter(int64(conf.VisitorSubscriptionLimit)),
 		seen:          time.Now(),
 	}