Add firebase subscriber to topics from cache upon initialization; stricter rate limits

.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.0/ntfy_1.2.0_amd64.deb
+dpkg -i ntfy_1.2.0_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.0/ntfy_1.2.0_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.0/ntfy_1.2.0_linux_x86_64.tar.gz
+sudo tar -C /usr/bin -zxf ntfy_1.2.0_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,7 +20,8 @@ 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.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: "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"}),
 	}
@@ -45,7 +46,8 @@ 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")
 
@@ -56,14 +58,15 @@ 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
 	s, err := server.New(conf)

config/config.go

@@ -8,39 +8,49 @@ import (
 
 // 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 the max number of requests, here:
-// 50 requests bucket, replenished at a rate of 1 per second
+// Defines all the limits
+// - request limit: max number of PUT/GET/.. requests (here: 50 requests bucket, replenished at a rate of one per 10 seconds)
+// - global topic limit: max number of topics overall
+// - subscription limit: max number of subscriptions (active HTTP connections) per per-visitor/IP
 var (
-	defaultLimit      = rate.Every(time.Second)
-	defaultLimitBurst = 50
+	defaultGlobalTopicLimit         = 5000
+	defaultVisitorRequestLimit      = rate.Every(10 * time.Second)
+	defaultVisitorRequestLimitBurst = 50
+	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
-	Limit                 rate.Limit
-	LimitBurst            int
+	ListenHTTP               string
+	FirebaseKeyFile          string
+	CacheFile                string
+	CacheDuration            time.Duration
+	KeepaliveInterval        time.Duration
+	ManagerInterval          time.Duration
+	GlobalTopicLimit         int
+	VisitorRequestLimit      rate.Limit
+	VisitorRequestLimitBurst int
+	VisitorSubscriptionLimit int
 }
 
 // New instantiates a default new config
 func New(listenHTTP string) *Config {
 	return &Config{
-		ListenHTTP:            listenHTTP,
-		FirebaseKeyFile:       "",
-		MessageBufferDuration: DefaultMessageBufferDuration,
-		KeepaliveInterval:     DefaultKeepaliveInterval,
-		ManagerInterval:       DefaultManagerInterval,
-		Limit:                 defaultLimit,
-		LimitBurst:            defaultLimitBurst,
+		ListenHTTP:               listenHTTP,
+		FirebaseKeyFile:          "",
+		CacheFile:                "",
+		CacheDuration:            DefaultCacheDuration,
+		KeepaliveInterval:        DefaultKeepaliveInterval,
+		ManagerInterval:          DefaultManagerInterval,
+		GlobalTopicLimit:         defaultGlobalTopicLimit,
+		VisitorRequestLimit:      defaultVisitorRequestLimit,
+		VisitorRequestLimitBurst: defaultVisitorRequestLimitBurst,
+		VisitorSubscriptionLimit: defaultVisitorSubscriptionLimit,
 	}
 }

config/config.yml

@@ -10,10 +10,15 @@
 #
 # 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.

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,7 +33,7 @@
     <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 allows you to send <b>notifications to your phone or desktop 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>
     <p id="error"></p>
@@ -81,6 +81,12 @@
     <ul id="topicsList"></ul>
     <audio id="notifySound" src="static/sound/mixkit-message-pop-alert-2354.mp3"></audio>
 
+    <h3>Subscribe via phone</h3>
+    <p>
+        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>
     <p class="smallMarginBottom">
         Using <a href="https://developer.mozilla.org/en-US/docs/Web/API/EventSource">EventSource</a> in JS, you can consume
@@ -142,6 +148,7 @@
         $ curl -s "ntfy.sh/mytopic/json?poll=1&since=10m"<br/>
         # Returns messages from up to 10 minutes ago and ends the connection
     </code>
+
     <h2>FAQ</h2>
     <p>
         <b>Isn't this like ...?</b><br/>
@@ -165,6 +172,28 @@
         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>
 
+    <p>
+        <b>Why is Firebase used?</b><br/>
+        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.
+    </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 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 service restarts.
+    </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>

server/server.go

@@ -9,7 +9,6 @@ import (
 	firebase "firebase.google.com/go"
 	"firebase.google.com/go/messaging"
 	"fmt"
-	"golang.org/x/time/rate"
 	"google.golang.org/api/option"
 	"heckel.io/ntfy/config"
 	"io"
@@ -23,9 +22,8 @@ import (
 	"time"
 )
 
-// TODO add "max connections open" limit
 // TODO add "max messages in a topic" limit
-// TODO add "max topics" limit
+// TODO implement "since=<ID>"
 
 // Server is the main server
 type Server struct {
@@ -34,15 +32,10 @@ type Server struct {
 	visitors map[string]*visitor
 	firebase subscriber
 	messages int64
+	cache    cache
 	mu       sync.Mutex
 }
 
-// visitor represents an API user, and its associated rate.Limiter used for rate limiting
-type visitor struct {
-	limiter *rate.Limiter
-	seen    time.Time
-}
-
 // errHTTP is a generic HTTP error for any non-200 HTTP error
 type errHTTP struct {
 	Code   int
@@ -54,8 +47,7 @@ func (e errHTTP) Error() string {
 }
 
 const (
-	messageLimit        = 1024
-	visitorExpungeAfter = 30 * time.Minute
+	messageLimit = 512
 )
 
 var (
@@ -86,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 {
@@ -147,21 +160,21 @@ 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 !v.limiter.Allow() {
-		return errHTTPTooManyRequests
+	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.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)
+		return s.handlePublish(w, r, v)
 	} else if r.Method == http.MethodGet && jsonRegex.MatchString(r.URL.Path) {
-		return s.handleSubscribeJSON(w, r)
+		return s.handleSubscribeJSON(w, r, v)
 	} else if r.Method == http.MethodGet && sseRegex.MatchString(r.URL.Path) {
-		return s.handleSubscribeSSE(w, r)
+		return s.handleSubscribeSSE(w, r, v)
 	} else if r.Method == http.MethodGet && rawRegex.MatchString(r.URL.Path) {
-		return s.handleSubscribeRaw(w, r)
+		return s.handleSubscribeRaw(w, r, v)
 	} else if r.Method == http.MethodOptions {
 		return s.handleOptions(w, r)
 	}
@@ -178,24 +191,34 @@ func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request) error {
 	return nil
 }
 
-func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request) error {
-	t := s.createTopic(r.URL.Path[1:])
+func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request, v *visitor) error {
+	t, err := s.topic(r.URL.Path[1:])
+	if err != nil {
+		return err
+	}
 	reader := io.LimitReader(r.Body, messageLimit)
 	b, err := io.ReadAll(reader)
 	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()
 	return nil
 }
 
-func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request) error {
+func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request, v *visitor) error {
 	encoder := func(msg *message) (string, error) {
 		var buf bytes.Buffer
 		if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
@@ -203,10 +226,10 @@ func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request) err
 		}
 		return buf.String(), nil
 	}
-	return s.handleSubscribe(w, r, "json", "application/stream+json", encoder)
+	return s.handleSubscribe(w, r, v, "json", "application/stream+json", encoder)
 }
 
-func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request) error {
+func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request, v *visitor) error {
 	encoder := func(msg *message) (string, error) {
 		var buf bytes.Buffer
 		if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
@@ -217,21 +240,28 @@ func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request) erro
 		}
 		return fmt.Sprintf("data: %s\n", buf.String()), nil
 	}
-	return s.handleSubscribe(w, r, "sse", "text/event-stream", encoder)
+	return s.handleSubscribe(w, r, v, "sse", "text/event-stream", encoder)
 }
 
-func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request) error {
+func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request, v *visitor) error {
 	encoder := func(msg *message) (string, error) {
-		if msg.Event == "" { // only handle default events
+		if msg.Event == messageEvent { // only handle default events
 			return strings.ReplaceAll(msg.Message, "\n", " ") + "\n", nil
 		}
 		return "\n", nil // "keepalive" and "open" events just send an empty line
 	}
-	return s.handleSubscribe(w, r, "raw", "text/plain", encoder)
+	return s.handleSubscribe(w, r, v, "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
+func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, v *visitor, format string, contentType string, encoder messageEncoder) error {
+	if err := v.AddSubscription(); err != nil {
+		return errHTTPTooManyRequests
+	}
+	defer v.RemoveSubscription()
+	t, err := s.topic(strings.TrimSuffix(r.URL.Path[1:], "/"+format)) // Hack
+	if err != nil {
+		return err
+	}
 	since, err := parseSince(r)
 	if err != nil {
 		return err
@@ -253,23 +283,22 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, format
 	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):
+			v.Keepalive()
 			if err := sub(newKeepaliveMessage(t.id)); err != nil { // Send keepalive message
 				return err
 			}
@@ -277,11 +306,15 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, format
 	}
 }
 
-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
 		}
@@ -308,16 +341,19 @@ func (s *Server) handleOptions(w http.ResponseWriter, r *http.Request) error {
 	return nil
 }
 
-func (s *Server) createTopic(id string) *topic {
+func (s *Server) topic(id string) (*topic, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if _, ok := s.topics[id]; !ok {
-		s.topics[id] = newTopic(id)
+		if len(s.topics) >= s.config.GlobalTopicLimit {
+			return nil, errHTTPTooManyRequests
+		}
+		s.topics[id] = newTopic(id, time.Now())
 		if s.firebase != nil {
 			s.topics[id].Subscribe(s.firebase)
 		}
 	}
-	return s.topics[id]
+	return s.topics[id], nil
 }
 
 func (s *Server) updateStatsAndExpire() {
@@ -326,27 +362,34 @@ func (s *Server) updateStatsAndExpire() {
 
 	// Expire visitors from rate visitors map
 	for ip, v := range s.visitors {
-		if time.Since(v.seen) > visitorExpungeAfter {
+		if v.Stale() {
 			delete(s.visitors, ip)
 		}
 	}
 
-	// Prune old messages, remove topics 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)) {
-			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))
 }
@@ -362,12 +405,8 @@ func (s *Server) visitor(remoteAddr string) *visitor {
 	}
 	v, exists := s.visitors[ip]
 	if !exists {
-		v = &visitor{
-			rate.NewLimiter(s.config.Limit, s.config.LimitBurst),
-			time.Now(),
-		}
-		s.visitors[ip] = v
-		return v
+		s.visitors[ip] = newVisitor(s.config)
+		return s.visitors[ip]
 	}
 	v.seen = time.Now()
 	return v

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

@@ -0,0 +1,65 @@
+package server
+
+import (
+	"golang.org/x/time/rate"
+	"heckel.io/ntfy/config"
+	"heckel.io/ntfy/util"
+	"sync"
+	"time"
+)
+
+const (
+	visitorExpungeAfter = 30 * time.Minute
+)
+
+// visitor represents an API user, and its associated rate.Limiter used for rate limiting
+type visitor struct {
+	config        *config.Config
+	limiter       *rate.Limiter
+	subscriptions *util.Limiter
+	seen          time.Time
+	mu            sync.Mutex
+}
+
+func newVisitor(conf *config.Config) *visitor {
+	return &visitor{
+		config:        conf,
+		limiter:       rate.NewLimiter(conf.VisitorRequestLimit, conf.VisitorRequestLimitBurst),
+		subscriptions: util.NewLimiter(int64(conf.VisitorSubscriptionLimit)),
+		seen:          time.Now(),
+	}
+}
+
+func (v *visitor) RequestAllowed() error {
+	if !v.limiter.Allow() {
+		return errHTTPTooManyRequests
+	}
+	return nil
+}
+
+func (v *visitor) AddSubscription() error {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	if err := v.subscriptions.Add(1); err != nil {
+		return errHTTPTooManyRequests
+	}
+	return nil
+}
+
+func (v *visitor) RemoveSubscription() {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	v.subscriptions.Sub(1)
+}
+
+func (v *visitor) Keepalive() {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	v.seen = time.Now()
+}
+
+func (v *visitor) Stale() bool {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	return time.Since(v.seen) > visitorExpungeAfter
+}

util/limit.go

@@ -0,0 +1,65 @@
+package util
+
+import (
+	"errors"
+	"sync"
+)
+
+// ErrLimitReached is the error returned by the Limiter and LimitWriter when the predefined limit has been reached
+var ErrLimitReached = errors.New("limit reached")
+
+// Limiter is a helper that allows adding values up to a well-defined limit. Once the limit is reached
+// ErrLimitReached will be returned. Limiter may be used by multiple goroutines.
+type Limiter struct {
+	value int64
+	limit int64
+	mu    sync.Mutex
+}
+
+// NewLimiter creates a new Limiter
+func NewLimiter(limit int64) *Limiter {
+	return &Limiter{
+		limit: limit,
+	}
+}
+
+// Add adds n to the limiters internal value, but only if the limit has not been reached. If the limit would be
+// exceeded after adding n, ErrLimitReached is returned.
+func (l *Limiter) Add(n int64) error {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.limit == 0 {
+		l.value += n
+		return nil
+	} else if l.value+n <= l.limit {
+		l.value += n
+		return nil
+	} else {
+		return ErrLimitReached
+	}
+}
+
+// Sub subtracts a value from the limiters internal value
+func (l *Limiter) Sub(n int64) {
+	l.Add(-n)
+}
+
+// Set sets the value of the limiter to n. This function ignores the limit. It is meant to set the value
+// based on reality.
+func (l *Limiter) Set(n int64) {
+	l.mu.Lock()
+	l.value = n
+	l.mu.Unlock()
+}
+
+// Value returns the internal value of the limiter
+func (l *Limiter) Value() int64 {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.value
+}
+
+// Limit returns the defined limit
+func (l *Limiter) Limit() int64 {
+	return l.limit
+}