chore: remove pubsub and introduce session manager

cmd/pg-ionscale-events/main.go

@@ -1,60 +0,0 @@
-package main
-
-import (
-	"database/sql"
-	"fmt"
-	"github.com/muesli/coral"
-	"os"
-	"time"
-
-	"github.com/lib/pq"
-)
-
-func main() {
-	cmd := rootCommand()
-	if err := cmd.Execute(); err != nil {
-		os.Exit(1)
-	}
-}
-
-func rootCommand() *coral.Command {
-	command := &coral.Command{
-		Use: "pg-ionscale-events",
-	}
-
-	var url string
-	command.Flags().StringVar(&url, "url", "", "")
-	_ = command.MarkFlagRequired("url")
-
-	command.RunE = func(cmd *coral.Command, args []string) error {
-		_, err := sql.Open("postgres", url)
-		if err != nil {
-			return err
-		}
-
-		reportProblem := func(ev pq.ListenerEventType, err error) {
-			if err != nil {
-				fmt.Println(err.Error())
-			}
-		}
-
-		minReconn := 10 * time.Second
-		maxReconn := time.Minute
-		listener := pq.NewListener(url, minReconn, maxReconn, reportProblem)
-		err = listener.Listen("ionscale_events")
-		if err != nil {
-			return err
-		}
-
-		fmt.Println("listening for events ...")
-		fmt.Println("")
-		for {
-			select {
-			case n, _ := <-listener.Notify:
-				fmt.Println(n.Extra)
-			}
-		}
-	}
-
-	return command
-}

go.mod

@@ -11,7 +11,6 @@ require (
 	github.com/glebarez/sqlite v1.5.0
 	github.com/go-gormigrate/gormigrate/v2 v2.0.2
 	github.com/golang-jwt/jwt/v4 v4.4.2
-	github.com/google/uuid v1.3.0
 	github.com/hashicorp/go-bexpr v0.1.11
 	github.com/hashicorp/go-hclog v1.3.0
 	github.com/hashicorp/go-multierror v1.1.1
@@ -20,7 +19,6 @@ require (
 	github.com/klauspost/compress v1.15.9
 	github.com/labstack/echo-contrib v0.13.0
 	github.com/labstack/echo/v4 v4.9.0
-	github.com/lib/pq v1.10.6
 	github.com/libdns/azure v0.2.0
 	github.com/libdns/cloudflare v0.1.0
 	github.com/libdns/digitalocean v0.0.0-20220518195853-a541bc8aa80f
@@ -90,6 +88,7 @@ require (
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/google/go-querystring v1.0.0 // indirect
+	github.com/google/uuid v1.3.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.0.0-20220520183353-fd19c99a87aa // indirect
 	github.com/googleapis/gax-go/v2 v2.4.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
@@ -109,6 +108,7 @@ require (
 	github.com/jsimonetti/rtnetlink v1.2.2 // indirect
 	github.com/klauspost/cpuid/v2 v2.1.1 // indirect
 	github.com/labstack/gommon v0.4.0 // indirect
+	github.com/lib/pq v1.10.6 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.16 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect

go.sum

@@ -58,6 +58,7 @@ cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9
 cloud.google.com/go/storage v1.22.0/go.mod h1:GbaLEoMqbVm6sx3Z0R++gSiBlgMv6yUi2q1DeGFKQgE=
 cloud.google.com/go/storage v1.22.1/go.mod h1:S8N1cAStu7BOeFfE8KAQzmyyLkK8p/vmRq6kuBTW58Y=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
+filippo.io/edwards25519 v1.0.0-rc.1 h1:m0VOOB23frXZvAOK44usCgLWvtsxIoMCTBGJZlpmGfU=
 filippo.io/mkcert v1.4.3 h1:axpnmtrZMM8u5Hf4N3UXxboGemMOV+Tn+e+pkHM6E3o=
 github.com/Azure/azure-sdk-for-go v52.4.0+incompatible h1:NpkT8MjJJMcgPJ5Q9E66QUgY9QRyxqM8MFx2P29uQZ4=
 github.com/Azure/azure-sdk-for-go v52.4.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
@@ -205,19 +206,15 @@ github.com/form3tech-oss/jwt-go v3.2.2+incompatible h1:TcekIExNqud5crz4xD2pavyTg
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/frankban/quicktest v1.14.0 h1:+cqqvzZV87b4adx/5ayVOaYZ2CrvM4ejQvUdBzPPUss=
 github.com/frankban/quicktest v1.14.0/go.mod h1:NeW+ay9A/U67EYXNFA1nPE8e/tnQv/09mUdL/ijj8og=
+github.com/fxamacker/cbor/v2 v2.4.0 h1:ri0ArlOR+5XunOP8CRUowT0pSJOwhW098ZCUyskZD88=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
 github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
 github.com/gin-gonic/gin v1.6.3 h1:ahKqKTFpO5KTPHxWZjEdPScmYaGtLo8Y4DMHoEsnp14=
 github.com/gin-gonic/gin v1.6.3/go.mod h1:75u5sXoLsGZoRN5Sgbi1eraJ4GU3++wFwWzhwvtwp4M=
-github.com/glebarez/go-sqlite v1.17.3/go.mod h1:Hg+PQuhUy98XCxWEJEaWob8x7lhJzhNYF1nZbUiRGIY=
-github.com/glebarez/go-sqlite v1.18.1 h1:w0xtxKWktqYsUsXg//SQK+l1IcpKb3rGOQHmMptvL2U=
-github.com/glebarez/go-sqlite v1.18.1/go.mod h1:ydXIGq2M4OzF4YyNhH129SPp7jWoVvgkEgb6pldmS0s=
 github.com/glebarez/go-sqlite v1.19.1/go.mod h1:9AykawGIyIcxoSfpYWiX1SgTNHTNsa/FVc75cDkbp4M=
 github.com/glebarez/go-sqlite v1.19.2 h1:mTtntWN3wk9UNjIf6F7Upqnfq96p+cjhfgCsupUd1hY=
 github.com/glebarez/go-sqlite v1.19.2/go.mod h1:DoubC3Kn5X6EBvDa2iaxAdIJqPNmY7M/sOCpfa8fus0=
-github.com/glebarez/sqlite v1.4.6 h1:D5uxD2f6UJ82cHnVtO2TZ9pqsLyto3fpDKHIk2OsR8A=
-github.com/glebarez/sqlite v1.4.6/go.mod h1:WYEtEFjhADPaPJqL/PGlbQQGINBA3eUAfDNbKFJf/zA=
 github.com/glebarez/sqlite v1.5.0 h1:+8LAEpmywqresSoGlqjjT+I9m4PseIM3NcerIJ/V7mk=
 github.com/glebarez/sqlite v1.5.0/go.mod h1:0wzXzTvfVJIN2GqRhCdMbnYd+m+aH5/QV7B30rM6NgY=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
@@ -362,6 +359,7 @@ github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+l
 github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 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/hdevalence/ed25519consensus v0.0.0-20220222234857-c00d1f31bab3 h1:aSVUgRRRtOrZOC1fYmY9gV0e9z/Iu+xNVSASWjsuyGU=
 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/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=
@@ -369,7 +367,6 @@ github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/inconshreveable/mousetrap v1.0.1 h1:U3uMjPSQEBMNp1lFxmllqCPM6P5u/Xq7Pgzkat/bFNc=
 github.com/inconshreveable/mousetrap v1.0.1/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
-github.com/jackc/chunkreader v1.0.0 h1:4s39bBR8ByfqH+DKm8rQA3E1LHZWB9XWcrz8fqaZbe0=
 github.com/jackc/chunkreader v1.0.0/go.mod h1:RT6O25fNZIuasFJRyZ4R/Y2BbhasbmZXF9QQ7T3kePo=
 github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
 github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8=
@@ -380,7 +377,6 @@ github.com/jackc/pgconn v0.0.0-20190831204454-2fabfa3c18b7/go.mod h1:ZJKsE/KZfsU
 github.com/jackc/pgconn v1.8.0/go.mod h1:1C2Pb36bGIP9QHGBYCjnyhqu7Rv3sGshaQUvmfGIB/o=
 github.com/jackc/pgconn v1.9.0/go.mod h1:YctiPyvzfU11JFxoXokUOOKQXQmDMoJL9vJzHH8/2JY=
 github.com/jackc/pgconn v1.9.1-0.20210724152538-d89c8390a530/go.mod h1:4z2w8XhRbP1hYxkpTuBjTS3ne3J48K83+u0zoyvg2pI=
-github.com/jackc/pgconn v1.12.1/go.mod h1:ZkhRC59Llhrq3oSfrikvwQ5NaxYExr6twkdkMLaKono=
 github.com/jackc/pgconn v1.13.0 h1:3L1XMNV2Zvca/8BYhzcRFS70Lr0WlDg16Di6SFGAbys=
 github.com/jackc/pgconn v1.13.0/go.mod h1:AnowpAqO4CMIIJNZl2VJp+KrkAZciAkhEl0W0JIobpI=
 github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE=
@@ -391,7 +387,6 @@ github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65 h1:DadwsjnMwFjfWc9y5W
 github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65/go.mod h1:5R2h2EEX+qri8jOWMbJCtaPWkrrNc7OHwsp2TCqp7ak=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
-github.com/jackc/pgproto3 v1.1.0 h1:FYYE4yRw+AgI8wXIinMlNjBbp/UitDJwfj5LqqewP1A=
 github.com/jackc/pgproto3 v1.1.0/go.mod h1:eR5FA3leWg7p9aeAqi37XOTgTIbkABlvcPB3E5rlc78=
 github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190420180111-c116219b62db/go.mod h1:bhq50y+xrl9n5mRYyCBFKkpRVTLYJVWeCc+mEAI3yXA=
 github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190609003834-432c2951c711/go.mod h1:uH0AWtUmuShn0bcesswc4aBTWGvw0cAxIJp+6OB//Wg=
@@ -399,7 +394,6 @@ github.com/jackc/pgproto3/v2 v2.0.0-rc3/go.mod h1:ryONWYqW6dqSg1Lw6vXNMXoBJhpzvW
 github.com/jackc/pgproto3/v2 v2.0.0-rc3.0.20190831210041-4c03ce451f29/go.mod h1:ryONWYqW6dqSg1Lw6vXNMXoBJhpzvWKnT95C46ckYeM=
 github.com/jackc/pgproto3/v2 v2.0.6/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
 github.com/jackc/pgproto3/v2 v2.1.1/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
-github.com/jackc/pgproto3/v2 v2.3.0/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
 github.com/jackc/pgproto3/v2 v2.3.1 h1:nwj7qwf0S+Q7ISFfBndqeLwSwxs+4DPsbRFjECT1Y4Y=
 github.com/jackc/pgproto3/v2 v2.3.1/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
 github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b h1:C8S2+VttkHFdOOCXJe+YGfa4vHYwlt4Zx+IVXQ97jYg=
@@ -408,20 +402,17 @@ github.com/jackc/pgtype v0.0.0-20190421001408-4ed0de4755e0/go.mod h1:hdSHsc1V01C
 github.com/jackc/pgtype v0.0.0-20190824184912-ab885b375b90/go.mod h1:KcahbBH1nCMSo2DXpzsoWOAfFkdEtEJpPbVLq8eE+mc=
 github.com/jackc/pgtype v0.0.0-20190828014616-a8802b16cc59/go.mod h1:MWlu30kVJrUS8lot6TQqcg7mtthZ9T0EoIBFiJcmcyw=
 github.com/jackc/pgtype v1.8.1-0.20210724151600-32e20a603178/go.mod h1:C516IlIV9NKqfsMCXTdChteoXmwgUceqaLfjg2e3NlM=
-github.com/jackc/pgtype v1.11.0/go.mod h1:LUMuVrfsFfdKGLw+AFFVv6KtHOFMwRgDDzBt76IqCA4=
 github.com/jackc/pgtype v1.12.0 h1:Dlq8Qvcch7kiehm8wPGIW0W3KsCCHJnRacKW0UM8n5w=
 github.com/jackc/pgtype v1.12.0/go.mod h1:LUMuVrfsFfdKGLw+AFFVv6KtHOFMwRgDDzBt76IqCA4=
 github.com/jackc/pgx/v4 v4.0.0-20190420224344-cc3461e65d96/go.mod h1:mdxmSJJuR08CZQyj1PVQBHy9XOp5p8/SHH6a0psbY9Y=
 github.com/jackc/pgx/v4 v4.0.0-20190421002000-1b8f0016e912/go.mod h1:no/Y67Jkk/9WuGR0JG/JseM9irFbnEPbuWV2EELPNuM=
 github.com/jackc/pgx/v4 v4.0.0-pre1.0.20190824185557-6972a5742186/go.mod h1:X+GQnOEnf1dqHGpw7JmHqHc1NxDoalibchSk9/RWuDc=
 github.com/jackc/pgx/v4 v4.12.1-0.20210724153913-640aa07df17c/go.mod h1:1QD0+tgSXP7iUjYm9C1NxKhny7lq6ee99u/z+IHFcgs=
-github.com/jackc/pgx/v4 v4.16.1/go.mod h1:SIhx0D5hoADaiXZVyv+3gSm3LCIIINTVO0PficsvWGQ=
 github.com/jackc/pgx/v4 v4.17.2 h1:0Ut0rpeKwvIVbMQ1KbMBU4h6wxehBI535LK6Flheh8E=
 github.com/jackc/pgx/v4 v4.17.2/go.mod h1:lcxIZN44yMIrWI78a5CpucdD14hX0SBDbNRvjDBItsw=
 github.com/jackc/puddle v0.0.0-20190413234325-e4ced69a3a2b/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v0.0.0-20190608224051-11cab39313c9/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
-github.com/jackc/puddle v1.2.1/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v1.3.0/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
 github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
@@ -497,8 +488,6 @@ github.com/libdns/libdns v0.2.1 h1:Wu59T7wSHRgtA0cfxC+n1c/e+O3upJGWytknkmFEDis=
 github.com/libdns/libdns v0.2.1/go.mod h1:yQCXzk1lEZmmCPa857bnk4TsOiqYasqpyOEeSObbb40=
 github.com/libdns/route53 v1.2.2 h1:ZnlxO2w8ftO/aR0PNRRB8lrG6AcKPOl/H0vU8mb/Ixo=
 github.com/libdns/route53 v1.2.2/go.mod h1:Vu827KwORxYR2I6iGsu8IKh4MESliECL7VA4pAsn95o=
-github.com/lxn/walk v0.0.0-20210112085537-c389da54e794/go.mod h1:E23UucZGqpuUANJooIbHWCufXvOcT6E7Stq81gU+CSQ=
-github.com/lxn/win v0.0.0-20210218163916-a377121e959e/go.mod h1:KxxjdtRkfNoYDCUP5ryK7XJJNTnpC8atvtmTheChOtk=
 github.com/mattn/go-colorable v0.1.1/go.mod h1:FuOcm+DKB9mbwrcAfNl7/TZVBZ6rcnceauSikq3lYCQ=
 github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
@@ -514,9 +503,7 @@ github.com/mattn/go-isatty v0.0.16 h1:bq3VjFmv/sOjHtdEhmkEV4x1AJtvUvOJ2PFAZ5+peK
 github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
 github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
 github.com/mattn/go-runewidth v0.0.13 h1:lTGmDsbAYt5DmK6OnoV7EuIF1wEIFAcxld6ypU4OSgU=
-github.com/mattn/go-sqlite3 v1.14.12/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
+github.com/mattn/go-sqlite3 v1.14.15 h1:vfoHhTN1af61xCRSWzFIWzx2YskyMTwHLrExkBOjvxI=
-github.com/mattn/go-sqlite3 v1.14.14 h1:qZgc/Rwetq+MtyE18WhzjokPD93dNqLGNT3QJuLvBGw=
-github.com/mattn/go-sqlite3 v1.14.14/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
 github.com/mattn/go-sqlite3 v1.14.15/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
@@ -584,7 +571,6 @@ github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1
 github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/prometheus/procfs v0.8.0 h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5mo=
 github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4=
-github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 h1:OdAsTTz6OkFY5QxjkYwrChwuRruF69c169dPK26NUlk=
 github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/remyoudompheng/bigfft v0.0.0-20220927061507-ef77025ab5aa h1:tEkEyxYeZ43TR55QU/hsIt9aRGBxbgGuz9CGykjvogY=
 github.com/remyoudompheng/bigfft v0.0.0-20220927061507-ef77025ab5aa/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
@@ -638,6 +624,7 @@ github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6Kllzaw
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 github.com/valyala/fasttemplate v1.2.1 h1:TVEnxayobAdVkhQfrfes2IzOB6o+z4roRkPF52WA1u4=
 github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
+github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/xhit/go-str2duration/v2 v2.0.0 h1:uFtk6FWB375bP7ewQl+/1wBcn840GPhnySOdcz/okPE=
 github.com/xhit/go-str2duration/v2 v2.0.0/go.mod h1:ohY8p+0f07DiV6Em5LKB0s2YpLtXVyJfNt1+BlmyAsU=
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@@ -697,14 +684,9 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201203163018-be400aefbc4c/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
 golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
-golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20221005025214-4161e89ecf1b h1:huxqepDufQpLLIRXiVkTvnxrzJlpwmIWAObmcCcUFr0=
-golang.org/x/crypto v0.0.0-20221005025214-4161e89ecf1b/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU=
 golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -786,7 +768,6 @@ golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96b
 golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20210903162142-ad29c8ab022f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
@@ -798,8 +779,6 @@ golang.org/x/net v0.0.0-20220607020251-c690dde0001d/go.mod h1:XRhObCWvk6IyKnWLug
 golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220630215102-69896b714898/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220826154423-83b083e8dc8b/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
-golang.org/x/net v0.0.0-20221004154528-8021a29435af h1:wv66FM3rLZGPdxpYL+ApnDe2HzHcTFta3z5nsc13wI4=
-golang.org/x/net v0.0.0-20221004154528-8021a29435af/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
 golang.org/x/net v0.1.0 h1:hZ/3BUoy5aId7sCpA/Tc5lt8DkFgdVS2onTpJsZ/fl0=
 golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -877,7 +856,6 @@ golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200803210538-64077c9b5642/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200905004654-be1d3432aa8f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20201018230417-eeed37f84f13/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201201145000-ef89a241ccb3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210104204734-6f8348627aad/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -888,7 +866,6 @@ golang.org/x/sys v0.0.0-20210305230114-8fe3ee5dd75b/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210315160823-c6e025ad8005/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210403161142-5e06dd20ab57/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210514084401-e8d321eab015/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -912,7 +889,6 @@ golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220405052023-b1e9470b6e64/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220408201424-a24fb2fb8a0f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220502124256-b6088ccd6cba/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -922,8 +898,6 @@ golang.org/x/sys v0.0.0-20220610221304-9f5ed59c137d/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20221010170243-090e33056c14 h1:k5II8e6QD8mITdi+okbbmR/cIyEbeXLBhy5Ha4nevyc=
-golang.org/x/sys v0.0.0-20221010170243-090e33056c14/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
@@ -938,8 +912,6 @@ 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/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.3.8-0.20211105212822-18b340fc7af2 h1:GLw7MR8AfAG2GmGcmVgObFOHXYypgGjnGno25RDwn3Y=
-golang.org/x/text v0.3.8-0.20211105212822-18b340fc7af2/go.mod h1:EFNZuWvGYxIRUEX+K8UmCFwYmZjqcrnq15ZuVldZkZ0=
 golang.org/x/text v0.4.0 h1:BrVqGRd7+k1DiOgtnFvAkoQEWQvBc25ouMJM6429SFg=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -1018,9 +990,6 @@ golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20220517211312-f3a8303e98df/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
 golang.org/x/xerrors v0.0.0-20220609144429-65e65417b02f h1:uF6paiQQebLeSXkrTqHqz0MXhXXS1KgF41eUdBNvxK0=
 golang.org/x/xerrors v0.0.0-20220609144429-65e65417b02f/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
-golang.zx2c4.com/wireguard v0.0.0-20210905140043-2ef39d47540c/go.mod h1:laHzsbfMhGSobUmruXWAyMKKHSqvIcrqZJMyHD+/3O8=
-golang.zx2c4.com/wireguard/windows v0.4.10 h1:HmjzJnb+G4NCdX+sfjsQlsxGPuYaThxRbZUZFLyR0/s=
-golang.zx2c4.com/wireguard/windows v0.4.10/go.mod h1:v7w/8FC48tTBm1IzScDVPEEb0/GjLta+T0ybpP9UWRg=
 golang.zx2c4.com/wireguard/windows v0.5.3 h1:On6j2Rpn3OEMXqBq00QEDC7bWSZrPIHKIus8eIuExIE=
 golang.zx2c4.com/wireguard/windows v0.5.3/go.mod h1:9TEe8TJmtwyQebdFwAkEWOPr3prrtqm+REGFifP60hI=
 google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE=
@@ -1227,15 +1196,11 @@ gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gorm.io/driver/mysql v1.3.3 h1:jXG9ANrwBc4+bMvBcSl8zCfPBaVoPyBEBshA8dA93X8=
-gorm.io/driver/postgres v1.3.9 h1:lWGiVt5CijhQAg0PWB7Od1RNcBw/jS4d2cAScBcSDXg=
-gorm.io/driver/postgres v1.3.9/go.mod h1:qw/FeqjxmYqW5dBcYNBsnhQULIApQdk7YuuDPktVi1U=
 gorm.io/driver/postgres v1.4.4 h1:zt1fxJ+C+ajparn0SteEnkoPg0BQ6wOWXEQ99bteAmw=
 gorm.io/driver/postgres v1.4.4/go.mod h1:whNfh5WhhHs96honoLjBAMwJGYEuA3m1hvgUbNXhPCw=
 gorm.io/driver/sqlite v1.3.2 h1:nWTy4cE52K6nnMhv23wLmur9Y3qWbZvOBz+V4PrGAxg=
 gorm.io/driver/sqlserver v1.3.2 h1:yYt8f/xdAKLY7lCCyXxIUEgZ/WsURos3dHrx8MKFGAk=
 gorm.io/gorm v1.23.7/go.mod h1:l2lP/RyAtc1ynaTjFksBde/O8v9oOGIApu2/xRitmZk=
-gorm.io/gorm v1.23.8 h1:h8sGJ+biDgBA1AD1Ha9gFCx7h8npU7AsLdlkX0n2TpE=
-gorm.io/gorm v1.23.8/go.mod h1:l2lP/RyAtc1ynaTjFksBde/O8v9oOGIApu2/xRitmZk=
 gorm.io/gorm v1.24.0 h1:j/CoiSm6xpRpmzbFJsQHYj+I8bGYWLXVHeYEyyKlF74=
 gorm.io/gorm v1.24.0/go.mod h1:DVrVomtaYTbqs7gB/x2uVvqnXzv0nqjB396B8cG4dBA=
 gotest.tools/v3 v3.0.2 h1:kG1BFyqVHuQoVQiR1bWGnfz/fmHvvuiSPIV7rvl360E=
@@ -1253,32 +1218,18 @@ inet.af/netaddr v0.0.0-20220811202034-502d2d690317 h1:U2fwK6P2EqmopP/hFLTOAjWTki
 inet.af/netaddr v0.0.0-20220811202034-502d2d690317/go.mod h1:OIezDfdzOgFhuw4HuWapWq2e9l0H9tK4F1j+ETRtF3k=
 lukechampine.com/uint128 v1.1.1/go.mod h1:c4eWIwlEGaxC/+H1VguhU4PHXNWDCDMUlWdIWl2j1gk=
 lukechampine.com/uint128 v1.2.0/go.mod h1:c4eWIwlEGaxC/+H1VguhU4PHXNWDCDMUlWdIWl2j1gk=
-modernc.org/cc/v3 v3.36.0/go.mod h1:NFUHyPn4ekoC/JHeZFfZurN6ixxawE1BnVonP/oahEI=
 modernc.org/cc/v3 v3.36.2/go.mod h1:NFUHyPn4ekoC/JHeZFfZurN6ixxawE1BnVonP/oahEI=
 modernc.org/cc/v3 v3.37.0/go.mod h1:vtL+3mdHx/wcj3iEGz84rQa8vEqR6XM84v5Lcvfph20=
 modernc.org/cc/v3 v3.38.1/go.mod h1:vtL+3mdHx/wcj3iEGz84rQa8vEqR6XM84v5Lcvfph20=
 modernc.org/cc/v3 v3.40.0/go.mod h1:/bTg4dnWkSXowUO6ssQKnOV0yMVxDYNIsIrzqTFDGH0=
-modernc.org/ccgo/v3 v3.0.0-20220428102840-41399a37e894/go.mod h1:eI31LL8EwEBKPpNpA4bU1/i+sKOwOrQy8D87zWUcRZc=
-modernc.org/ccgo/v3 v3.0.0-20220430103911-bc99d88307be/go.mod h1:bwdAnOoaIt8Ax9YdWGjxWsdkPcZyRPHqrOvJxaKAKGw=
 modernc.org/ccgo/v3 v3.0.0-20220904174949-82d86e1b6d56/go.mod h1:YSXjPL62P2AMSxBphRHPn7IkzhVHqkvOnRKAKh+W6ZI=
 modernc.org/ccgo/v3 v3.0.0-20220910160915-348f15de615a/go.mod h1:8p47QxPkdugex9J4n9P2tLZ9bK01yngIVp00g4nomW0=
-modernc.org/ccgo/v3 v3.16.4/go.mod h1:tGtX0gE9Jn7hdZFeU88slbTh1UtCYKusWOoCJuvkWsQ=
-modernc.org/ccgo/v3 v3.16.6/go.mod h1:tGtX0gE9Jn7hdZFeU88slbTh1UtCYKusWOoCJuvkWsQ=
-modernc.org/ccgo/v3 v3.16.8/go.mod h1:zNjwkizS+fIFDrDjIAgBSCLkWbJuHF+ar3QRn+Z9aws=
 modernc.org/ccgo/v3 v3.16.9/go.mod h1:zNMzC9A9xeNUepy6KuZBbugn3c0Mc9TeiJO4lgvkJDo=
 modernc.org/ccgo/v3 v3.16.12/go.mod h1:fUB3Vn0nVPReA+7IG7yZDfjv1TMWjhQP8gCxrFAtL5g=
 modernc.org/ccorpus v1.11.6/go.mod h1:2gEUTrWqdpH2pXsmTM1ZkjeSrUWDpjMu2T6m29L/ErQ=
 modernc.org/httpfs v1.0.6/go.mod h1:7dosgurJGp0sPaRanU53W4xZYKh14wfzX420oZADeHM=
-modernc.org/libc v0.0.0-20220428101251-2d5f3daf273b/go.mod h1:p7Mg4+koNjc8jkqwcoFBJx7tXkpj00G77X7A72jXPXA=
-modernc.org/libc v1.16.0/go.mod h1:N4LD6DBE9cf+Dzf9buBlzVJndKr/iJHG97vGLHYnb5A=
-modernc.org/libc v1.16.1/go.mod h1:JjJE0eu4yeK7tab2n4S1w8tlWd9MxXLRzheaRnAKymU=
-modernc.org/libc v1.16.7/go.mod h1:hYIV5VZczAmGZAnG15Vdngn5HSF5cSkbvfz2B7GRuVU=
-modernc.org/libc v1.16.8/go.mod h1:hYIV5VZczAmGZAnG15Vdngn5HSF5cSkbvfz2B7GRuVU=
-modernc.org/libc v1.16.17/go.mod h1:hYIV5VZczAmGZAnG15Vdngn5HSF5cSkbvfz2B7GRuVU=
-modernc.org/libc v1.16.19/go.mod h1:p7Mg4+koNjc8jkqwcoFBJx7tXkpj00G77X7A72jXPXA=
 modernc.org/libc v1.17.0/go.mod h1:XsgLldpP4aWlPlsjqKRdHPqCxCjISdHfM/yeWC5GyW0=
 modernc.org/libc v1.17.4/go.mod h1:WNg2ZH56rDEwdropAJeZPQkXmDwh+JCA1s/htl6r2fA=
-modernc.org/libc v1.18.0 h1:EKpC8eyhOcxpstYjohs7vxni7BoQBUVWXsf5rAZzlgk=
 modernc.org/libc v1.18.0/go.mod h1:vj6zehR5bfc98ipowQOM2nIDUZnVew/wNC/2tOGS+q0=
 modernc.org/libc v1.19.0/go.mod h1:ZRfIaEkgrYgZDl6pa4W39HgN5G/yDW+NRmNKZBDFrk0=
 modernc.org/libc v1.20.3/go.mod h1:ZRfIaEkgrYgZDl6pa4W39HgN5G/yDW+NRmNKZBDFrk0=
@@ -1288,28 +1239,21 @@ modernc.org/mathutil v1.2.2/go.mod h1:mZW8CKdRPY1v87qxC/wUdX5O1qDzXMP5TH3wjfpga6
 modernc.org/mathutil v1.4.1/go.mod h1:mZW8CKdRPY1v87qxC/wUdX5O1qDzXMP5TH3wjfpga6E=
 modernc.org/mathutil v1.5.0 h1:rV0Ko/6SfM+8G+yKiyI830l3Wuz1zRutdslNoQ0kfiQ=
 modernc.org/mathutil v1.5.0/go.mod h1:mZW8CKdRPY1v87qxC/wUdX5O1qDzXMP5TH3wjfpga6E=
-modernc.org/memory v1.1.1/go.mod h1:/0wo5ibyrQiaoUoH7f9D8dnglAmILJ5/cxZlRECf+Nw=
 modernc.org/memory v1.2.0/go.mod h1:/0wo5ibyrQiaoUoH7f9D8dnglAmILJ5/cxZlRECf+Nw=
-modernc.org/memory v1.3.0 h1:6ZIOLb5ronARPxEPxtZz1WbSRllgA09FCvNNyql5kZg=
 modernc.org/memory v1.3.0/go.mod h1:PkUhL0Mugw21sHPeskwZW4D6VscE/GQJOnIpCnW6pSU=
 modernc.org/memory v1.4.0 h1:crykUfNSnMAXaOJnnxcSzbUGMqkLWjklJKkBK2nwZwk=
 modernc.org/memory v1.4.0/go.mod h1:PkUhL0Mugw21sHPeskwZW4D6VscE/GQJOnIpCnW6pSU=
 modernc.org/opt v0.1.1/go.mod h1:WdSiB5evDcignE70guQKxYUl14mgWtbClRi5wmkkTX0=
 modernc.org/opt v0.1.3/go.mod h1:WdSiB5evDcignE70guQKxYUl14mgWtbClRi5wmkkTX0=
-modernc.org/sqlite v1.17.3/go.mod h1:10hPVYar9C0kfXuTWGz8s0XtB8uAGymUy51ZzStYe3k=
-modernc.org/sqlite v1.18.1 h1:ko32eKt3jf7eqIkCgPAeHMBXw3riNSLhl2f3loEF7o8=
-modernc.org/sqlite v1.18.1/go.mod h1:6ho+Gow7oX5V+OiOQ6Tr4xeqbx13UZ6t+Fw9IRUG4d4=
 modernc.org/sqlite v1.19.1/go.mod h1:UfQ83woKMaPW/ZBruK0T7YaFCrI+IE0LeWVY6pmnVms=
 modernc.org/sqlite v1.19.2 h1:1VaNHEe6amuHhelmAOtibvYpAjwLfT4q6cBB2K7ZlQ8=
 modernc.org/sqlite v1.19.2/go.mod h1:fEgebDYAGTFJj2c/ukKmnaq/0ZQZg0PSYxRa/bHyCDs=
 modernc.org/strutil v1.1.1/go.mod h1:DE+MQQ/hjKBZS2zNInV5hhcipt5rLPWkmpbGeW5mmdw=
 modernc.org/strutil v1.1.3/go.mod h1:MEHNA7PdEnEwLvspRMtWTNnp2nnyvMfkimT1NKNAGbw=
-modernc.org/tcl v1.13.1/go.mod h1:XOLfOwzhkljL4itZkK6T72ckMgvj0BDsnKNdZVUOecw=
 modernc.org/tcl v1.14.0/go.mod h1:gQ7c1YPMvryCHCcmf8acB6VPabE59QBeuRQLL7cTUlM=
 modernc.org/tcl v1.15.0/go.mod h1:xRoGotBZ6dU+Zo2tca+2EqVEeMmOUBzHnhIwq4YrVnE=
 modernc.org/token v1.0.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
 modernc.org/token v1.0.1/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
-modernc.org/z v1.5.1/go.mod h1:eWFB510QWW5Th9YGZT81s+LwvaAs3Q2yr4sP0rmLkv8=
 modernc.org/z v1.6.0/go.mod h1:hVdgNMh8ggTuRG1rGU8x+xGRFfiQUIAw0ZqlPy8+HyQ=
 modernc.org/z v1.7.0/go.mod h1:hVdgNMh8ggTuRG1rGU8x+xGRFfiQUIAw0ZqlPy8+HyQ=
 nhooyr.io/websocket v1.8.7 h1:usjR2uOr/zjjkVMy0lW+PPohFok7PCow5sDjLgX4P4g=
@@ -1318,7 +1262,5 @@ rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8
 rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
 rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=
 software.sslmate.com/src/go-pkcs12 v0.0.0-20210415151418-c5206de65a78 h1:SqYE5+A2qvRhErbsXFfUEUmpWEKxxRSMgGLkvRAFOV4=
-tailscale.com v1.30.2 h1:aOoRbHmtIW/uZoP1o9YRyilffW4Nv9sqF6nQr7GWXzg=
-tailscale.com v1.30.2/go.mod h1:MO+tWkQp2YIF3KBnnej/mQvgYccRS5Xk/IrEpZ4Z3BU=
 tailscale.com v1.32.0 h1:00V30ztnQMTvQA9YV+bg28WamrGmUzWlc5SfWSlXz9c=
 tailscale.com v1.32.0/go.mod h1:AwKk+tI7z1+EXzmLxVNXA4dBpgTKd6RVV1VjFYEiv94=

internal/broker/broker.go

@@ -1 +0,0 @@
-package broker

internal/broker/pubsub.go

@@ -1,16 +0,0 @@
-package broker
-
-type Signal struct {
-	PeerUpdated  *uint64
-	PeersRemoved []uint64
-	ACLUpdated   bool
-	DNSUpdated   bool
-}
-
-type Listener chan *Signal
-
-type Pubsub interface {
-	Subscribe(tailnet uint64, listener Listener) (cancel func(), err error)
-	Publish(tailnet uint64, message *Signal) error
-	Close() error
-}

internal/broker/pubsub_mem.go

@@ -1,61 +0,0 @@
-package broker
-
-import (
-	"github.com/google/uuid"
-	"sync"
-)
-
-type memoryPubsub struct {
-	mut       sync.RWMutex
-	listeners map[uint64]map[uuid.UUID]Listener
-}
-
-func (m *memoryPubsub) Subscribe(tailnet uint64, listener Listener) (cancel func(), err error) {
-	m.mut.Lock()
-	defer m.mut.Unlock()
-
-	var listeners map[uuid.UUID]Listener
-	var ok bool
-	if listeners, ok = m.listeners[tailnet]; !ok {
-		listeners = map[uuid.UUID]Listener{}
-		m.listeners[tailnet] = listeners
-	}
-	var id uuid.UUID
-	for {
-		id = uuid.New()
-		if _, ok = listeners[id]; !ok {
-			break
-		}
-	}
-
-	listeners[id] = listener
-	return func() {
-		m.mut.Lock()
-		defer m.mut.Unlock()
-		listeners := m.listeners[tailnet]
-		delete(listeners, id)
-	}, nil
-}
-
-func (m *memoryPubsub) Publish(tailnet uint64, message *Signal) error {
-	m.mut.RLock()
-	defer m.mut.RUnlock()
-	listeners, ok := m.listeners[tailnet]
-	if !ok {
-		return nil
-	}
-	for _, listener := range listeners {
-		listener <- message
-	}
-	return nil
-}
-
-func (*memoryPubsub) Close() error {
-	return nil
-}
-
-func NewPubsubInMemory() Pubsub {
-	return &memoryPubsub{
-		listeners: make(map[uint64]map[uuid.UUID]Listener),
-	}
-}

internal/broker/pubsub_postgres.go

@@ -1,117 +0,0 @@
-package broker
-
-import (
-	"context"
-	"database/sql"
-	"encoding/json"
-	"fmt"
-	"github.com/lib/pq"
-	"time"
-)
-
-type pgPubsub struct {
-	pgListener *pq.Listener
-	db         *sql.DB
-	target     Pubsub
-}
-
-func NewPubsub(ctx context.Context, database *sql.DB, connectURL string) (Pubsub, error) {
-	errCh := make(chan error)
-	listener := pq.NewListener(connectURL, time.Second, time.Minute, func(event pq.ListenerEventType, err error) {
-		select {
-		case <-errCh:
-			return
-		default:
-			errCh <- err
-			close(errCh)
-		}
-	})
-
-	select {
-	case err := <-errCh:
-		if err != nil {
-			return nil, fmt.Errorf("create pq listener: %w", err)
-		}
-	case <-ctx.Done():
-		return nil, ctx.Err()
-	}
-
-	if err := listener.Listen("ionscale_events"); err != nil {
-		return nil, fmt.Errorf("listen: %w", err)
-	}
-
-	pubsub := &pgPubsub{
-		db:         database,
-		pgListener: listener,
-		target:     NewPubsubInMemory(),
-	}
-	go pubsub.listen(ctx)
-
-	return pubsub, nil
-}
-
-func (p *pgPubsub) Close() error {
-	return p.pgListener.Close()
-}
-
-func (p *pgPubsub) Subscribe(tailnet uint64, listener Listener) (cancel func(), err error) {
-	return p.target.Subscribe(tailnet, listener)
-}
-
-func (p *pgPubsub) Publish(tailnet uint64, message *Signal) error {
-	event := &pgEvent{
-		TailnetID: tailnet,
-		Signal:    message,
-	}
-
-	payload, err := json.Marshal(event)
-	if err != nil {
-		return err
-	}
-
-	_, err = p.db.ExecContext(context.Background(), `select pg_notify(`+pq.QuoteLiteral("ionscale_events")+`, $1)`, payload)
-	if err != nil {
-		return fmt.Errorf("exec pg_notify: %w", err)
-	}
-
-	return nil
-}
-
-func (p *pgPubsub) listen(ctx context.Context) {
-	var (
-		notif *pq.Notification
-		ok    bool
-	)
-	defer p.pgListener.Close()
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case notif, ok = <-p.pgListener.Notify:
-			if !ok {
-				return
-			}
-		}
-		// A nil notification can be dispatched on reconnect.
-		if notif == nil {
-			continue
-		}
-		p.listenReceive(notif)
-	}
-}
-
-func (p *pgPubsub) listenReceive(notif *pq.Notification) {
-	extra := []byte(notif.Extra)
-	event := &pgEvent{}
-
-	if err := json.Unmarshal(extra, event); err == nil {
-		p.target.Publish(event.TailnetID, event.Signal)
-	} else {
-		fmt.Println(err)
-	}
-}
-
-type pgEvent struct {
-	TailnetID uint64
-	Signal    *Signal
-}

internal/core/reaper.go

@@ -1,8 +1,7 @@
-package handlers
+package core
 
 import (
 	"context"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"time"
 )
@@ -12,26 +11,28 @@ const (
 	inactivityTimeout = 30 * time.Minute
 )
 
-func NewReaper(brokers broker.Pubsub, repository domain.Repository) *Reaper {
+func StartReaper(repository domain.Repository, sessionManager PollMapSessionManager) {
-	return &Reaper{
+	r := &reaper{
-		pubsub:     brokers,
+		sessionManager: sessionManager,
-		repository: repository,
+		repository:     repository,
 	}
+
+	go r.start()
 }
 
-type Reaper struct {
+type reaper struct {
-	pubsub     broker.Pubsub
+	sessionManager PollMapSessionManager
-	repository domain.Repository
+	repository     domain.Repository
 }
 
-func (r *Reaper) Start() {
+func (r *reaper) start() {
 	t := time.NewTicker(ticker)
 	for range t.C {
 		r.reapInactiveEphemeralNodes()
 	}
 }
 
-func (r *Reaper) reapInactiveEphemeralNodes() {
+func (r *reaper) reapInactiveEphemeralNodes() {
 	ctx := context.Background()
 
 	now := time.Now().UTC()
@@ -54,8 +55,8 @@ func (r *Reaper) reapInactiveEphemeralNodes() {
 	}
 
 	if len(removedNodes) != 0 {
-		for i, p := range removedNodes {
+		for i, _ := range removedNodes {
-			r.pubsub.Publish(i, &broker.Signal{PeersRemoved: p})
+			r.sessionManager.NotifyAll(i)
 		}
 	}
 }

internal/core/session_manager.go

@@ -0,0 +1,94 @@
+package core
+
+import (
+	"sync"
+	"time"
+)
+
+type Ping struct{}
+
+type PollMapSessionManager interface {
+	Register(tailnetID uint64, machineID uint64, ch chan *Ping)
+	Deregister(tailnetID uint64, machineID uint64)
+	HasSession(tailnetID uint64, machineID uint64) bool
+	NotifyAll(tailnetID uint64)
+}
+
+func NewPollMapSessionManager() PollMapSessionManager {
+	return &pollMapSessionManager{
+		data:   map[uint64]map[uint64]chan *Ping{},
+		timers: map[uint64]*time.Timer{},
+	}
+}
+
+type pollMapSessionManager struct {
+	sync.RWMutex
+	data   map[uint64]map[uint64]chan *Ping
+	timers map[uint64]*time.Timer
+}
+
+func (n *pollMapSessionManager) Register(tailnetID uint64, machineID uint64, ch chan *Ping) {
+	n.Lock()
+	defer n.Unlock()
+
+	if ss := n.data[tailnetID]; ss == nil {
+		n.data[tailnetID] = map[uint64]chan *Ping{machineID: ch}
+	} else {
+		ss[machineID] = ch
+	}
+
+	t, ok := n.timers[machineID]
+	if ok {
+		t.Stop()
+		delete(n.timers, machineID)
+	}
+}
+
+func (n *pollMapSessionManager) Deregister(tailnetID uint64, machineID uint64) {
+	n.Lock()
+	defer n.Unlock()
+
+	if ss := n.data[tailnetID]; ss != nil {
+		delete(ss, machineID)
+	}
+
+	t, ok := n.timers[machineID]
+	if ok {
+		t.Stop()
+		delete(n.timers, machineID)
+	}
+
+	timer := time.NewTimer(10 * time.Second)
+	go func() {
+		<-timer.C
+		if !n.HasSession(tailnetID, machineID) {
+			n.NotifyAll(tailnetID)
+		}
+	}()
+
+	n.timers[machineID] = timer
+}
+
+func (n *pollMapSessionManager) HasSession(tailnetID uint64, machineID uint64) bool {
+	n.RLock()
+	defer n.RUnlock()
+
+	if ss := n.data[tailnetID]; ss != nil {
+		if _, ok := ss[machineID]; ok {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (n *pollMapSessionManager) NotifyAll(tailnetID uint64) {
+	n.RLock()
+	defer n.RUnlock()
+
+	if ss := n.data[tailnetID]; ss != nil {
+		for _, p := range ss {
+			p <- &Ping{}
+		}
+	}
+}

internal/database/database.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"github.com/go-gormigrate/gormigrate/v2"
 	"github.com/hashicorp/go-hclog"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/database/migration"
 	"github.com/jsiebens/ionscale/internal/util"
 	"tailscale.com/types/key"
@@ -25,51 +24,38 @@ type db interface {
 	UnlockErr(error) error
 }
 
-func OpenDB(config *config.Database, logger hclog.Logger) (domain.Repository, broker.Pubsub, error) {
+func OpenDB(config *config.Database, logger hclog.Logger) (domain.Repository, error) {
-	db, pubsub, err := createDB(config, logger)
+	db, err := createDB(config, logger)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	repository := domain.NewRepository(db.DB())
 
 	if err := db.Lock(); err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	if err := db.UnlockErr(migrate(db.DB())); err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
-	return repository, pubsub, nil
+	return repository, nil
 }
 
-func createDB(config *config.Database, logger hclog.Logger) (db, broker.Pubsub, error) {
+func createDB(config *config.Database, logger hclog.Logger) (db, error) {
 	gormConfig := &gorm.Config{
 		Logger: &GormLoggerAdapter{logger: logger.Named("db")},
 	}
 
 	switch config.Type {
 	case "sqlite", "sqlite3":
-		db, err := newSqliteDB(config, gormConfig)
+		return newSqliteDB(config, gormConfig)
-		return db, broker.NewPubsubInMemory(), err
 	case "postgres", "postgresql":
-		db, err := newPostgresDB(config, gormConfig)
+		return newPostgresDB(config, gormConfig)
-		if err != nil {
-			return nil, nil, err
-		}
-		stdDB, err := db.DB().DB()
-		if err != nil {
-			return nil, nil, err
-		}
-		pubsub, err := broker.NewPubsub(context.TODO(), stdDB, config.Url)
-		if err != nil {
-			return nil, nil, err
-		}
-		return db, pubsub, err
 	}
 
-	return nil, nil, fmt.Errorf("invalid database type '%s'", config.Type)
+	return nil, fmt.Errorf("invalid database type '%s'", config.Type)
 }
 
 func migrate(db *gorm.DB) error {

internal/handlers/poll_net_map.go

@@ -3,8 +3,8 @@ package handlers
 import (
 	"context"
 	"github.com/jsiebens/ionscale/internal/bind"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/config"
+	"github.com/jsiebens/ionscale/internal/core"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"github.com/jsiebens/ionscale/internal/errors"
 	"github.com/jsiebens/ionscale/internal/mapping"
@@ -17,25 +17,22 @@ import (
 
 func NewPollNetMapHandler(
 	createBinder bind.Factory,
-	brokers broker.Pubsub,
+	sessionManager core.PollMapSessionManager,
-	repository domain.Repository,
+	repository domain.Repository) *PollNetMapHandler {
-	offlineTimers *OfflineTimers) *PollNetMapHandler {
 
 	handler := &PollNetMapHandler{
-		createBinder:  createBinder,
+		createBinder:   createBinder,
-		brokers:       brokers,
+		sessionManager: sessionManager,
-		repository:    repository,
+		repository:     repository,
-		offlineTimers: offlineTimers,
 	}
 
 	return handler
 }
 
 type PollNetMapHandler struct {
-	createBinder  bind.Factory
+	createBinder   bind.Factory
-	repository    domain.Repository
+	repository     domain.Repository
-	brokers       broker.Pubsub
+	sessionManager core.PollMapSessionManager
-	offlineTimers *OfflineTimers
 }
 
 func (h *PollNetMapHandler) PollNetMap(c echo.Context) error {
@@ -87,7 +84,7 @@ func (h *PollNetMapHandler) handleUpdate(c echo.Context, binder bind.Binder, m *
 	tailnetID := m.TailnetID
 	machineID := m.ID
 
-	h.brokers.Publish(tailnetID, &broker.Signal{PeerUpdated: &machineID})
+	h.sessionManager.NotifyAll(tailnetID)
 
 	if !mapRequest.Stream {
 		return c.String(http.StatusOK, "")
@@ -101,13 +98,8 @@ func (h *PollNetMapHandler) handleUpdate(c echo.Context, binder bind.Binder, m *
 		return errors.Wrap(err, 0)
 	}
 
-	updateChan := make(chan *broker.Signal, 20)
+	updateChan := make(chan *core.Ping, 20)
-
+	h.sessionManager.Register(m.TailnetID, m.ID, updateChan)
-	unsubscribe, err := h.brokers.Subscribe(tailnetID, updateChan)
-	if err != nil {
-		return errors.Wrap(err, 0)
-	}
-	h.cancelOfflineMessage(machineID)
 
 	// Listen to connection close
 	notify := c.Request().Context().Done()
@@ -116,8 +108,6 @@ func (h *PollNetMapHandler) handleUpdate(c echo.Context, binder bind.Binder, m *
 	if err != nil {
 		return errors.Wrap(err, 0)
 	}
-	keepAliveTicker := time.NewTicker(config.KeepAliveInterval())
-	syncTicker := time.NewTicker(5 * time.Second)
 
 	c.Response().WriteHeader(http.StatusOK)
 
@@ -128,13 +118,15 @@ func (h *PollNetMapHandler) handleUpdate(c echo.Context, binder bind.Binder, m *
 
 	connectedDevices.WithLabelValues(m.Tailnet.Name).Inc()
 
+	keepAliveTicker := time.NewTicker(config.KeepAliveInterval())
+	syncTicker := time.NewTicker(5 * time.Second)
+
 	defer func() {
 		connectedDevices.WithLabelValues(m.Tailnet.Name).Dec()
-		unsubscribe()
+		h.sessionManager.Deregister(m.TailnetID, m.ID)
 		keepAliveTicker.Stop()
 		syncTicker.Stop()
 		_ = h.repository.SetMachineLastSeen(ctx, machineID)
-		h.scheduleOfflineMessage(tailnetID, machineID)
 	}()
 
 	var latestSync = time.Now()
@@ -203,14 +195,6 @@ func (h *PollNetMapHandler) handleReadOnly(c echo.Context, binder bind.Binder, m
 	return errors.Wrap(err, 0)
 }
 
-func (h *PollNetMapHandler) scheduleOfflineMessage(tailnetID, machineID uint64) {
-	h.offlineTimers.startCh <- [2]uint64{tailnetID, machineID}
-}
-
-func (h *PollNetMapHandler) cancelOfflineMessage(machineID uint64) {
-	h.offlineTimers.stopCh <- machineID
-}
-
 func (h *PollNetMapHandler) createKeepAliveResponse(binder bind.Binder, request *tailcfg.MapRequest) ([]byte, error) {
 	mapResponse := &tailcfg.MapResponse{
 		KeepAlive: true,
@@ -228,7 +212,7 @@ func (h *PollNetMapHandler) createMapResponse(m *domain.Machine, binder bind.Bin
 	}
 
 	hostinfo := tailcfg.Hostinfo(m.HostInfo)
-	node, user, err := mapping.ToNode(m, tailnet, false)
+	node, user, err := mapping.ToNode(m, tailnet, false, true)
 	if err != nil {
 		return nil, nil, "", err
 	}
@@ -251,7 +235,7 @@ func (h *PollNetMapHandler) createMapResponse(m *domain.Machine, binder bind.Bin
 			continue
 		}
 		if policies.IsValidPeer(m, &peer) || policies.IsValidPeer(&peer, m) {
-			n, u, err := mapping.ToNode(&peer, tailnet, true)
+			n, u, err := mapping.ToNode(&peer, tailnet, true, h.sessionManager.HasSession(peer.TailnetID, peer.ID))
 			if err != nil {
 				return nil, nil, "", err
 			}
@@ -331,60 +315,6 @@ func (h *PollNetMapHandler) createMapResponse(m *domain.Machine, binder bind.Bin
 	return payload, syncedPeerIDs, derpMap.Checksum, nil
 }
 
-func NewOfflineTimers(repository domain.Repository, pubsub broker.Pubsub) *OfflineTimers {
-	return &OfflineTimers{
-		repository: repository,
-		pubsub:     pubsub,
-		data:       make(map[uint64]*time.Timer),
-		startCh:    make(chan [2]uint64),
-		stopCh:     make(chan uint64),
-	}
-}
-
-type OfflineTimers struct {
-	repository domain.Repository
-	pubsub     broker.Pubsub
-	data       map[uint64]*time.Timer
-	stopCh     chan uint64
-	startCh    chan [2]uint64
-}
-
-func (o *OfflineTimers) Start() {
-	for {
-		select {
-		case i := <-o.startCh:
-			o.scheduleOfflineMessage(i[0], i[1])
-		case m := <-o.stopCh:
-			o.cancelOfflineMessage(m)
-		}
-	}
-}
-
-func (o *OfflineTimers) scheduleOfflineMessage(tailnetID, machineID uint64) {
-	t, ok := o.data[machineID]
-	if ok {
-		t.Stop()
-		delete(o.data, machineID)
-	}
-
-	timer := time.NewTimer(config.KeepAliveInterval())
-	go func() {
-		<-timer.C
-		o.pubsub.Publish(tailnetID, &broker.Signal{PeerUpdated: &machineID})
-		o.stopCh <- machineID
-	}()
-
-	o.data[machineID] = timer
-}
-
-func (o *OfflineTimers) cancelOfflineMessage(machineID uint64) {
-	t, ok := o.data[machineID]
-	if ok {
-		t.Stop()
-		delete(o.data, machineID)
-	}
-}
-
 func optBool(v bool) opt.Bool {
 	b := opt.Bool("")
 	b.Set(v)

internal/handlers/registration.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"github.com/jsiebens/ionscale/internal/addr"
 	"github.com/jsiebens/ionscale/internal/bind"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/config"
+	"github.com/jsiebens/ionscale/internal/core"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"github.com/jsiebens/ionscale/internal/errors"
 	"github.com/jsiebens/ionscale/internal/mapping"
@@ -21,21 +21,21 @@ import (
 func NewRegistrationHandlers(
 	createBinder bind.Factory,
 	config *config.Config,
-	brokers broker.Pubsub,
+	sessionManager core.PollMapSessionManager,
 	repository domain.Repository) *RegistrationHandlers {
 	return &RegistrationHandlers{
-		createBinder: createBinder,
+		createBinder:   createBinder,
-		pubsub:       brokers,
+		sessionManager: sessionManager,
-		repository:   repository,
+		repository:     repository,
-		config:       config,
+		config:         config,
 	}
 }
 
 type RegistrationHandlers struct {
-	createBinder bind.Factory
+	createBinder   bind.Factory
-	repository   domain.Repository
+	repository     domain.Repository
-	pubsub       broker.Pubsub
+	sessionManager core.PollMapSessionManager
-	config       *config.Config
+	config         *config.Config
 }
 
 func (h *RegistrationHandlers) Register(c echo.Context) error {
@@ -74,12 +74,12 @@ func (h *RegistrationHandlers) Register(c echo.Context) error {
 				if _, err := h.repository.DeleteMachine(ctx, m.ID); err != nil {
 					return errors.Wrap(err, 0)
 				}
-				h.pubsub.Publish(m.TailnetID, &broker.Signal{PeersRemoved: []uint64{m.ID}})
+				h.sessionManager.NotifyAll(m.TailnetID)
 			} else {
 				if err := h.repository.SaveMachine(ctx, m); err != nil {
 					return errors.Wrap(err, 0)
 				}
-				h.pubsub.Publish(m.TailnetID, &broker.Signal{PeerUpdated: &m.ID})
+				h.sessionManager.NotifyAll(m.TailnetID)
 			}
 
 			response := tailcfg.RegisterResponse{NodeKeyExpired: true}

internal/mapping/mapping.go

@@ -82,7 +82,7 @@ func ToDNSConfig(m *domain.Machine, tailnet *domain.Tailnet, c *domain.DNSConfig
 	return dnsConfig
 }
 
-func ToNode(m *domain.Machine, tailnet *domain.Tailnet, peer bool) (*tailcfg.Node, *tailcfg.UserProfile, error) {
+func ToNode(m *domain.Machine, tailnet *domain.Tailnet, peer bool, connected bool) (*tailcfg.Node, *tailcfg.UserProfile, error) {
 	role := tailnet.IAMPolicy.GetRole(m.User)
 
 	var capabilities []string
@@ -194,13 +194,9 @@ func ToNode(m *domain.Machine, tailnet *domain.Tailnet, peer bool) (*tailcfg.Nod
 		n.KeyExpiry = time.Time{}
 	}
 
-	if m.LastSeen != nil {
+	n.Online = &connected
-		l := m.LastSeen.UTC()
+	if !connected && m.LastSeen != nil {
-		online := m.LastSeen.After(time.Now().Add(-config.KeepAliveInterval()))
+		n.LastSeen = m.LastSeen
-		n.Online = &online
-		if !online {
-			n.LastSeen = &l
-		}
 	}
 
 	var user = ToUserProfile(m.User)

internal/server/server.go

@@ -9,6 +9,7 @@ import (
 	"github.com/jsiebens/ionscale/internal/auth"
 	"github.com/jsiebens/ionscale/internal/bind"
 	"github.com/jsiebens/ionscale/internal/config"
+	"github.com/jsiebens/ionscale/internal/core"
 	"github.com/jsiebens/ionscale/internal/database"
 	"github.com/jsiebens/ionscale/internal/dns"
 	"github.com/jsiebens/ionscale/internal/domain"
@@ -38,11 +39,13 @@ func Start(c *config.Config) error {
 
 	logger.Info("Starting ionscale server")
 
-	repository, brokers, err := database.OpenDB(&c.Database, logger)
+	repository, err := database.OpenDB(&c.Database, logger)
 	if err != nil {
 		return err
 	}
 
+	sessionManager := core.NewPollMapSessionManager()
+
 	defaultControlKeys, err := repository.GetControlKeys(context.Background())
 	if err != nil {
 		return err
@@ -53,11 +56,7 @@ func Start(c *config.Config) error {
 		return err
 	}
 
-	offlineTimers := handlers.NewOfflineTimers(repository, brokers)
+	core.StartReaper(repository, sessionManager)
-	reaper := handlers.NewReaper(brokers, repository)
-
-	go offlineTimers.Start()
-	go reaper.Start()
 
 	serverUrl, err := url.Parse(c.ServerUrl)
 	if err != nil {
@@ -100,8 +99,8 @@ func Start(c *config.Config) error {
 	createPeerHandler := func(machinePublicKey key.MachinePublic) http.Handler {
 		binder := bind.DefaultBinder(machinePublicKey)
 
-		registrationHandlers := handlers.NewRegistrationHandlers(binder, c, brokers, repository)
+		registrationHandlers := handlers.NewRegistrationHandlers(binder, c, sessionManager, repository)
-		pollNetMapHandler := handlers.NewPollNetMapHandler(binder, brokers, repository, offlineTimers)
+		pollNetMapHandler := handlers.NewPollNetMapHandler(binder, sessionManager, repository)
 		dnsHandlers := handlers.NewDNSHandlers(binder, dnsProvider)
 		idTokenHandlers := handlers.NewIDTokenHandlers(binder, c, repository)
 		sshActionHandlers := handlers.NewSSHActionHandlers(binder, c, repository)
@@ -119,8 +118,8 @@ func Start(c *config.Config) error {
 	}
 
 	noiseHandlers := handlers.NewNoiseHandlers(serverKey.ControlKey, createPeerHandler)
-	registrationHandlers := handlers.NewRegistrationHandlers(bind.BoxBinder(serverKey.LegacyControlKey), c, brokers, repository)
+	registrationHandlers := handlers.NewRegistrationHandlers(bind.BoxBinder(serverKey.LegacyControlKey), c, sessionManager, repository)
-	pollNetMapHandler := handlers.NewPollNetMapHandler(bind.BoxBinder(serverKey.LegacyControlKey), brokers, repository, offlineTimers)
+	pollNetMapHandler := handlers.NewPollNetMapHandler(bind.BoxBinder(serverKey.LegacyControlKey), sessionManager, repository)
 	dnsHandlers := handlers.NewDNSHandlers(bind.BoxBinder(serverKey.LegacyControlKey), dnsProvider)
 	idTokenHandlers := handlers.NewIDTokenHandlers(bind.BoxBinder(serverKey.LegacyControlKey), c, repository)
 	authenticationHandlers := handlers.NewAuthenticationHandlers(
@@ -130,7 +129,7 @@ func Start(c *config.Config) error {
 		repository,
 	)
 
-	rpcService := service.NewService(c, authProvider, repository, brokers)
+	rpcService := service.NewService(c, authProvider, repository, sessionManager)
 	rpcPath, rpcHandler := NewRpcHandler(serverKey.SystemAdminKey, repository, logger, rpcService)
 
 	nonTlsAppHandler := echo.New()

internal/service/acl.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"github.com/bufbuild/connect-go"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"github.com/jsiebens/ionscale/internal/errors"
 	"github.com/jsiebens/ionscale/internal/mapping"
@@ -57,7 +56,7 @@ func (s *Service) SetACLPolicy(ctx context.Context, req *connect.Request[api.Set
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(tailnet.ID, &broker.Signal{ACLUpdated: true})
+	s.sessionManager.NotifyAll(tailnet.ID)
 
 	return connect.NewResponse(&api.SetACLPolicyResponse{}), nil
 }

internal/service/derp_map.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/bufbuild/connect-go"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"github.com/jsiebens/ionscale/internal/errors"
 	"github.com/jsiebens/ionscale/internal/util"
@@ -58,7 +57,7 @@ func (s *Service) SetDefaultDERPMap(ctx context.Context, req *connect.Request[ap
 	}
 
 	for _, t := range tailnets {
-		s.pubsub.Publish(t.ID, &broker.Signal{})
+		s.sessionManager.NotifyAll(t.ID)
 	}
 
 	return connect.NewResponse(&api.SetDefaultDERPMapResponse{Value: req.Msg.Value}), nil
@@ -82,7 +81,7 @@ func (s *Service) ResetDefaultDERPMap(ctx context.Context, req *connect.Request[
 	}
 
 	for _, t := range tailnets {
-		s.pubsub.Publish(t.ID, &broker.Signal{})
+		s.sessionManager.NotifyAll(t.ID)
 	}
 
 	return connect.NewResponse(&api.ResetDefaultDERPMapResponse{}), nil

internal/service/dns.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"github.com/bufbuild/connect-go"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/config"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"github.com/jsiebens/ionscale/internal/errors"
@@ -74,7 +73,7 @@ func (s *Service) SetDNSConfig(ctx context.Context, req *connect.Request[api.Set
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(tailnet.ID, &broker.Signal{DNSUpdated: true})
+	s.sessionManager.NotifyAll(tailnet.ID)
 
 	resp := &api.SetDNSConfigResponse{Config: dnsConfig}
 

internal/service/machine.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"fmt"
 	"github.com/bufbuild/connect-go"
-	"github.com/jsiebens/ionscale/internal/broker"
-	"github.com/jsiebens/ionscale/internal/config"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"github.com/jsiebens/ionscale/internal/errors"
 	api "github.com/jsiebens/ionscale/pkg/gen/ionscale/v1"
@@ -22,10 +20,9 @@ func (s *Service) machineToApi(m *domain.Machine) *api.Machine {
 		name = fmt.Sprintf("%s-%d", m.Name, m.NameIdx)
 	}
 
-	online := false
+	online := s.sessionManager.HasSession(m.TailnetID, m.ID)
 	if m.LastSeen != nil {
 		lastSeen = timestamppb.New(*m.LastSeen)
-		online = m.LastSeen.After(time.Now().Add(-config.KeepAliveInterval()))
 	}
 
 	return &api.Machine{
@@ -127,7 +124,7 @@ func (s *Service) DeleteMachine(ctx context.Context, req *connect.Request[api.De
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(m.TailnetID, &broker.Signal{PeersRemoved: []uint64{m.ID}})
+	s.sessionManager.NotifyAll(m.TailnetID)
 
 	return connect.NewResponse(&api.DeleteMachineResponse{}), nil
 }
@@ -156,7 +153,7 @@ func (s *Service) ExpireMachine(ctx context.Context, req *connect.Request[api.Ex
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(m.TailnetID, &broker.Signal{PeerUpdated: &m.ID})
+	s.sessionManager.NotifyAll(m.TailnetID)
 
 	return connect.NewResponse(&api.ExpireMachineResponse{}), nil
 }
@@ -184,7 +181,7 @@ func (s *Service) AuthorizeMachine(ctx context.Context, req *connect.Request[api
 		}
 	}
 
-	s.pubsub.Publish(m.TailnetID, &broker.Signal{PeerUpdated: &m.ID})
+	s.sessionManager.NotifyAll(m.TailnetID)
 
 	return connect.NewResponse(&api.AuthorizeMachineResponse{}), nil
 }
@@ -256,7 +253,7 @@ func (s *Service) EnableMachineRoutes(ctx context.Context, req *connect.Request[
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(m.TailnetID, &broker.Signal{PeerUpdated: &m.ID})
+	s.sessionManager.NotifyAll(m.TailnetID)
 
 	response := api.EnableMachineRoutesResponse{
 		MachineId: m.ID,
@@ -305,7 +302,7 @@ func (s *Service) DisableMachineRoutes(ctx context.Context, req *connect.Request
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(m.TailnetID, &broker.Signal{PeerUpdated: &m.ID})
+	s.sessionManager.NotifyAll(m.TailnetID)
 
 	response := api.DisableMachineRoutesResponse{
 		MachineId: m.ID,
@@ -352,7 +349,7 @@ func (s *Service) EnableExitNode(ctx context.Context, req *connect.Request[api.E
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(m.TailnetID, &broker.Signal{PeerUpdated: &m.ID})
+	s.sessionManager.NotifyAll(m.TailnetID)
 
 	response := api.EnableExitNodeResponse{
 		MachineId: m.ID,
@@ -403,7 +400,7 @@ func (s *Service) DisableExitNode(ctx context.Context, req *connect.Request[api.
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(m.TailnetID, &broker.Signal{PeerUpdated: &m.ID})
+	s.sessionManager.NotifyAll(m.TailnetID)
 
 	response := api.DisableExitNodeResponse{
 		MachineId: m.ID,
@@ -440,7 +437,7 @@ func (s *Service) SetMachineKeyExpiry(ctx context.Context, req *connect.Request[
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(m.TailnetID, &broker.Signal{PeerUpdated: &m.ID})
+	s.sessionManager.NotifyAll(m.TailnetID)
 
 	return connect.NewResponse(&api.SetMachineKeyExpiryResponse{}), nil
 }

internal/service/service.go

@@ -4,27 +4,27 @@ import (
 	"context"
 	"github.com/bufbuild/connect-go"
 	"github.com/jsiebens/ionscale/internal/auth"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/config"
+	"github.com/jsiebens/ionscale/internal/core"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"github.com/jsiebens/ionscale/internal/version"
 	api "github.com/jsiebens/ionscale/pkg/gen/ionscale/v1"
 )
 
-func NewService(config *config.Config, authProvider auth.Provider, repository domain.Repository, pubsub broker.Pubsub) *Service {
+func NewService(config *config.Config, authProvider auth.Provider, repository domain.Repository, sessionManager core.PollMapSessionManager) *Service {
 	return &Service{
-		config:       config,
+		config:         config,
-		authProvider: authProvider,
+		authProvider:   authProvider,
-		repository:   repository,
+		repository:     repository,
-		pubsub:       pubsub,
+		sessionManager: sessionManager,
 	}
 }
 
 type Service struct {
-	config       *config.Config
+	config         *config.Config
-	authProvider auth.Provider
+	authProvider   auth.Provider
-	repository   domain.Repository
+	repository     domain.Repository
-	pubsub       broker.Pubsub
+	sessionManager core.PollMapSessionManager
 }
 
 func (s *Service) GetVersion(_ context.Context, _ *connect.Request[api.GetVersionRequest]) (*connect.Response[api.GetVersionResponse], error) {

internal/service/tailnet.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/bufbuild/connect-go"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"github.com/jsiebens/ionscale/internal/errors"
 	"github.com/jsiebens/ionscale/internal/util"
@@ -141,7 +140,7 @@ func (s *Service) DeleteTailnet(ctx context.Context, req *connect.Request[api.De
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(req.Msg.TailnetId, &broker.Signal{})
+	s.sessionManager.NotifyAll(req.Msg.TailnetId)
 
 	return connect.NewResponse(&api.DeleteTailnetResponse{}), nil
 }
@@ -174,7 +173,7 @@ func (s *Service) SetDERPMap(ctx context.Context, req *connect.Request[api.SetDE
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(tailnet.ID, &broker.Signal{})
+	s.sessionManager.NotifyAll(tailnet.ID)
 
 	raw, err := json.Marshal(derpMap)
 	if err != nil {
@@ -204,7 +203,7 @@ func (s *Service) ResetDERPMap(ctx context.Context, req *connect.Request[api.Res
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(tailnet.ID, &broker.Signal{})
+	s.sessionManager.NotifyAll(tailnet.ID)
 
 	return connect.NewResponse(&api.ResetDERPMapResponse{}), nil
 }
@@ -256,7 +255,7 @@ func (s *Service) EnableFileSharing(ctx context.Context, req *connect.Request[ap
 			return nil, errors.Wrap(err, 0)
 		}
 
-		s.pubsub.Publish(tailnet.ID, &broker.Signal{})
+		s.sessionManager.NotifyAll(tailnet.ID)
 	}
 
 	return connect.NewResponse(&api.EnableFileSharingResponse{}), nil
@@ -282,7 +281,7 @@ func (s *Service) DisableFileSharing(ctx context.Context, req *connect.Request[a
 			return nil, errors.Wrap(err, 0)
 		}
 
-		s.pubsub.Publish(tailnet.ID, &broker.Signal{})
+		s.sessionManager.NotifyAll(tailnet.ID)
 	}
 
 	return connect.NewResponse(&api.DisableFileSharingResponse{}), nil
@@ -308,7 +307,7 @@ func (s *Service) EnableServiceCollection(ctx context.Context, req *connect.Requ
 			return nil, errors.Wrap(err, 0)
 		}
 
-		s.pubsub.Publish(tailnet.ID, &broker.Signal{})
+		s.sessionManager.NotifyAll(tailnet.ID)
 	}
 
 	return connect.NewResponse(&api.EnableServiceCollectionResponse{}), nil
@@ -334,7 +333,7 @@ func (s *Service) DisableServiceCollection(ctx context.Context, req *connect.Req
 			return nil, errors.Wrap(err, 0)
 		}
 
-		s.pubsub.Publish(tailnet.ID, &broker.Signal{})
+		s.sessionManager.NotifyAll(tailnet.ID)
 	}
 
 	return connect.NewResponse(&api.DisableServiceCollectionResponse{}), nil
@@ -360,7 +359,7 @@ func (s *Service) EnableSSH(ctx context.Context, req *connect.Request[api.Enable
 			return nil, errors.Wrap(err, 0)
 		}
 
-		s.pubsub.Publish(tailnet.ID, &broker.Signal{})
+		s.sessionManager.NotifyAll(tailnet.ID)
 	}
 
 	return connect.NewResponse(&api.EnableSSHResponse{}), nil
@@ -386,7 +385,7 @@ func (s *Service) DisableSSH(ctx context.Context, req *connect.Request[api.Disab
 			return nil, errors.Wrap(err, 0)
 		}
 
-		s.pubsub.Publish(tailnet.ID, &broker.Signal{})
+		s.sessionManager.NotifyAll(tailnet.ID)
 	}
 
 	return connect.NewResponse(&api.DisableSSHResponse{}), nil

internal/service/users.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"github.com/bufbuild/connect-go"
-	"github.com/jsiebens/ionscale/internal/broker"
 	"github.com/jsiebens/ionscale/internal/domain"
 	"github.com/jsiebens/ionscale/internal/errors"
 	api "github.com/jsiebens/ionscale/pkg/gen/ionscale/v1"
@@ -91,7 +90,7 @@ func (s *Service) DeleteUser(ctx context.Context, req *connect.Request[api.Delet
 		return nil, errors.Wrap(err, 0)
 	}
 
-	s.pubsub.Publish(user.TailnetID, &broker.Signal{})
+	s.sessionManager.NotifyAll(user.TailnetID)
 
 	return connect.NewResponse(&api.DeleteUserResponse{}), nil
 }