amadan
Add Send and SendTo over webpush-go
webpush-go's types stay behind the package boundary. Results report acceptance per device and the batch error what stopped the batch, so a failed query never reads as zero devices. 404/410 prune and 2xx confirm both match the captured revision, under a context the caller's cancellation cannot interrupt: a 410 the service already answered must prune. Transport errors are stripped of the endpoint URL before they can reach a log. Concurrency is a semaphore across the Service with a WaitGroup, so Send returns only after every goroutine it started has stopped writing; a slot freed by a request the cancellation cut short cannot start another, because the context is re-checked after the slot is taken. Options.TTL zero means 24 hours — webpush-go always sends the header and a literal 0 means "deliver now or drop" — which is a deviation from the spec's "service default", recorded in the branch discussion. Endpoints are validated on the send path too, since a caller can edit a Stored.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
4 files changed,
+653
−4
go.mod+7 −2go.sum+70 −2send.go+234 −0send_test.go+342 −0
diff --git a/go.mod b/go.mod| index e513ce0..66c1a97 100644 |
| --- a/go.mod |
| +++ b/go.mod |
| @@ -2,18 +2,23 @@ module amadan.net/rastrillo/aviso |
| go 1.25.0 |
| -require amadan.net/rastrillo/rastrillo v0.26.0 |
| +require ( |
| + amadan.net/rastrillo/rastrillo v0.26.0 |
| + github.com/SherClockHolmes/webpush-go v1.4.0 |
| +) |
| require ( |
| github.com/dustin/go-humanize v1.0.1 // indirect |
| + github.com/golang-jwt/jwt/v5 v5.2.1 // indirect |
| github.com/google/uuid v1.6.0 // indirect |
| github.com/jinzhu/inflection v1.0.0 // indirect |
| github.com/jinzhu/now v1.1.5 // indirect |
| github.com/mattn/go-isatty v0.0.20 // indirect |
| github.com/ncruces/go-strftime v1.0.0 // indirect |
| github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect |
| + golang.org/x/crypto v0.31.0 // indirect |
| golang.org/x/sys v0.46.0 // indirect |
| - golang.org/x/text v0.20.0 // indirect |
| + golang.org/x/text v0.21.0 // indirect |
| gorm.io/gorm v1.31.2 // indirect |
| gorm.io/plugin/dbresolver v1.6.2 // indirect |
| modernc.org/libc v1.74.1 // indirect |
diff --git a/go.sum b/go.sum| index 2e62e67..7c585c7 100644 |
| --- a/go.sum |
| +++ b/go.sum |
| @@ -1,9 +1,14 @@ |
| amadan.net/rastrillo/rastrillo v0.26.0 h1:I7UkiDbT304q9wXnmgabe84P6RDFTjMbKU174i1QZDo= |
| amadan.net/rastrillo/rastrillo v0.26.0/go.mod h1:RpyHVPD0udcSfHJroY0KZE/FUjgw6Tvfo8SJm2qeZGE= |
| +github.com/SherClockHolmes/webpush-go v1.4.0 h1:ocnzNKWN23T9nvHi6IfyrQjkIc0oJWv1B1pULsf9i3s= |
| +github.com/SherClockHolmes/webpush-go v1.4.0/go.mod h1:XSq8pKX11vNV8MJEMwjrlTkxhAj1zKfxmyhdV7Pd6UA= |
| github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= |
| github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= |
| github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc= |
| github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI= |
| +github.com/golang-jwt/jwt/v5 v5.2.1 h1:OuVbFODueb089Lh128TAcimifWaLhJwVflnrgM17wHk= |
| +github.com/golang-jwt/jwt/v5 v5.2.1/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= |
| +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= |
| github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= |
| github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= |
| github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= |
| @@ -24,17 +29,80 @@ github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOF |
| github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= |
| github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= |
| github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= |
| +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= |
| +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= |
| +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= |
| +golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc= |
| +golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= |
| +golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8= |
| +golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U= |
| +golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= |
| +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= |
| +golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= |
| +golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= |
| +golang.org/x/mod v0.15.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= |
| +golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= |
| golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= |
| golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= |
| +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= |
| +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= |
| +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= |
| +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= |
| +golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= |
| +golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk= |
| +golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44= |
| +golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM= |
| +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= |
| +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= |
| +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= |
| +golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y= |
| +golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= |
| +golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= |
| +golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= |
| golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= |
| golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= |
| +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= |
| +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= |
| +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= |
| +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= |
| +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= |
| +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= |
| golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= |
| +golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= |
| +golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= |
| +golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= |
| +golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= |
| +golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= |
| golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= |
| golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= |
| -golang.org/x/text v0.20.0 h1:gK/Kv2otX8gz+wn7Rmb3vT96ZwuoxnQlY+HlJVj7Qug= |
| -golang.org/x/text v0.20.0/go.mod h1:D4IsuqiFMhST5bX19pQ9ikHC2GsaKyk/oF+pn3ducp4= |
| +golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE= |
| +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= |
| +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= |
| +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= |
| +golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo= |
| +golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU= |
| +golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk= |
| +golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY= |
| +golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM= |
| +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= |
| +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= |
| +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= |
| +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= |
| +golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= |
| +golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= |
| +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= |
| +golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= |
| +golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo= |
| +golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= |
| +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= |
| +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= |
| +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= |
| +golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= |
| +golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58= |
| +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= |
| golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= |
| golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= |
| +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= |
| gorm.io/driver/mysql v1.5.7 h1:MndhOPYOfEp2rHKgkZIhJ16eVUIRf2HmzgoPmh7FCWo= |
| gorm.io/driver/mysql v1.5.7/go.mod h1:sEtPWMiqiN1N1cMXoXmBbd8C6/l+TESwriotuRRpkDM= |
| gorm.io/driver/sqlite v1.6.0 h1:WHRRrIiulaPiPFmDcod6prc4l2VGVWHz80KspNsxSfQ= |
diff --git a/send.go b/send.go| new file mode 100644 |
| index 0000000..82e105f |
| --- /dev/null |
| +++ b/send.go |
| @@ -0,0 +1,234 @@ |
| +package aviso |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "net/http" |
| + "net/url" |
| + "strconv" |
| + "strings" |
| + "sync" |
| + "time" |
| + |
| + webpush "github.com/SherClockHolmes/webpush-go" |
| +) |
| + |
| +// Options tune one batch. Zero values mean aviso's defaults, not the |
| +// push service's: webpush-go always sends a TTL header, and a literal |
| +// 0 there means "deliver now or drop". |
| +type Options struct { |
| + // TTL is how long the push service may hold the message; whole |
| + // seconds, >= 0. 0 means 24 hours. |
| + TTL time.Duration |
| + // Urgency is "very-low", "low", "normal" or "high"; "" means normal. |
| + Urgency string |
| + // Topic collapses pending messages with the same topic; <= 32 |
| + // URL-safe characters; "" means none. |
| + Topic string |
| +} |
| + |
| +// Result is one device's outcome. Status is the push service's |
| +// acceptance, not delivery; a 2xx means the service took it. |
| +type Result struct { |
| + ID string |
| + Status int // 0 when Err is transport-level or the row was never attempted |
| + RetryAfter time.Duration // from a 429/503, else 0 |
| + Err error |
| +} |
| + |
| +// ErrPayloadTooLarge means the plaintext exceeds RFC 8291's one-record |
| +// limit; a larger payload would be split, which no browser accepts. |
| +var ErrPayloadTooLarge = errors.New("aviso: payload over 3993 bytes") |
| + |
| +// ErrBadOptions means Options failed validation. |
| +var ErrBadOptions = errors.New("aviso: invalid Options") |
| + |
| +const ( |
| + maxPayload = 3993 |
| + defaultTTL = 24 * time.Hour |
| + requestTimeout = 30 * time.Second |
| + maxBodyRead = 4096 |
| +) |
| + |
| +func (o Options) validate() error { |
| + if o.TTL < 0 || o.TTL%time.Second != 0 { |
| + return fmt.Errorf("%w: TTL must be whole non-negative seconds", ErrBadOptions) |
| + } |
| + switch o.Urgency { |
| + case "", "very-low", "low", "normal", "high": |
| + default: |
| + return fmt.Errorf("%w: Urgency %q", ErrBadOptions, o.Urgency) |
| + } |
| + if len(o.Topic) > 32 || strings.Trim(o.Topic, "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_") != "" { |
| + return fmt.Errorf("%w: Topic must be <= 32 URL-safe characters", ErrBadOptions) |
| + } |
| + return nil |
| +} |
| + |
| +// SendTo fans payload out to every device subject enrolled — the |
| +// common case, so an app never touches Stored. The batch error covers |
| +// what stops the batch (the query, validation, cancellation); each |
| +// Result covers one device. |
| +func (s *Service) SendTo(ctx context.Context, subject string, payload []byte, o Options) ([]Result, error) { |
| + if err := checkBatch(ctx, payload, o); err != nil { |
| + return nil, err |
| + } |
| + rows, err := s.List(ctx, subject) |
| + if err != nil { |
| + return nil, err |
| + } |
| + return s.Send(ctx, rows, payload, o) |
| +} |
| + |
| +// Send delivers payload to each of to, bounded by Config.Concurrency |
| +// across the Service. It never retries: RetryAfter is for the app's |
| +// own scheduler. |
| +// |
| +// On cancellation, rows not yet started carry ctx.Err(), rows already |
| +// in flight finish (their own request honours ctx), and Send returns |
| +// only once every goroutine it started has stopped writing results — |
| +// a caller that reads results after Send returns must never race a |
| +// straggler. |
| +func (s *Service) Send(ctx context.Context, to []Stored, payload []byte, o Options) ([]Result, error) { |
| + if err := checkBatch(ctx, payload, o); err != nil { |
| + return nil, err |
| + } |
| + results := make([]Result, len(to)) |
| + var wg sync.WaitGroup |
| + var batchErr error |
| + for i := range to { |
| + st := to[i] |
| + results[i].ID = st.ID |
| + if st.VAPIDKeyID != s.keyID { |
| + results[i].Err = ErrKeyMismatch |
| + continue |
| + } |
| + if err := validateEndpoint(st.Endpoint); err != nil { |
| + results[i].Err = err |
| + continue |
| + } |
| + if batchErr != nil { |
| + results[i].Err = batchErr |
| + continue |
| + } |
| + select { |
| + case s.sem <- struct{}{}: |
| + case <-ctx.Done(): |
| + batchErr = ctx.Err() |
| + results[i].Err = batchErr |
| + continue |
| + } |
| + // A slot freed by a request the cancellation cut short can win |
| + // the select above; nothing starts after cancellation. |
| + if err := ctx.Err(); err != nil { |
| + <-s.sem |
| + batchErr = err |
| + results[i].Err = err |
| + continue |
| + } |
| + wg.Add(1) |
| + go func(i int, st Stored) { |
| + defer wg.Done() |
| + defer func() { <-s.sem }() |
| + results[i] = s.sendOne(ctx, st, payload, o) |
| + }(i, st) |
| + } |
| + wg.Wait() |
| + if batchErr == nil && ctx.Err() != nil { |
| + batchErr = ctx.Err() |
| + } |
| + return results, batchErr |
| +} |
| + |
| +func checkBatch(ctx context.Context, payload []byte, o Options) error { |
| + if err := ctx.Err(); err != nil { |
| + return err |
| + } |
| + if len(payload) > maxPayload { |
| + return ErrPayloadTooLarge |
| + } |
| + return o.validate() |
| +} |
| + |
| +func (s *Service) sendOne(ctx context.Context, st Stored, payload []byte, o Options) Result { |
| + res := Result{ID: st.ID} |
| + ctx, cancel := context.WithTimeout(ctx, requestTimeout) |
| + defer cancel() |
| + ttl := o.TTL |
| + if ttl == 0 { |
| + ttl = defaultTTL |
| + } |
| + urgency := webpush.Urgency(o.Urgency) |
| + if urgency == "" { |
| + urgency = webpush.UrgencyNormal |
| + } |
| + resp, err := webpush.SendNotificationWithContext(ctx, payload, |
| + &webpush.Subscription{Endpoint: st.Endpoint, Keys: webpush.Keys{P256dh: st.P256dh, Auth: st.Auth}}, |
| + &webpush.Options{ |
| + HTTPClient: s.client, |
| + Subscriber: s.wireContact, |
| + TTL: int(ttl / time.Second), |
| + Urgency: urgency, |
| + Topic: o.Topic, |
| + VAPIDPublicKey: s.pub, |
| + VAPIDPrivateKey: s.cfg.PrivateKey, |
| + }) |
| + if err != nil { |
| + res.Err = redact(err) |
| + return res |
| + } |
| + defer resp.Body.Close() |
| + // Drain a bounded amount for keep-alive; the body is never read |
| + // into anything a log could see. |
| + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBodyRead)) |
| + res.Status = resp.StatusCode |
| + // Store updates run under a context the caller's cancellation |
| + // cannot interrupt: a 410 the service already answered must prune |
| + // whether or not the batch was cancelled a moment later. |
| + dbCtx := context.WithoutCancel(ctx) |
| + switch { |
| + case resp.StatusCode >= 200 && resp.StatusCode < 300: |
| + if err := s.confirm(dbCtx, st.ID, st.Revision); err != nil { |
| + s.cfg.Logger.Warn("aviso: confirm failed", "id", st.ID, "err", err) |
| + } |
| + case resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone: |
| + res.Err = fmt.Errorf("aviso: push service says subscription gone (%d)", resp.StatusCode) |
| + if err := s.prune(dbCtx, st.ID, st.Revision); err != nil { |
| + s.cfg.Logger.Warn("aviso: prune failed", "id", st.ID, "err", err) |
| + } |
| + case resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode == http.StatusServiceUnavailable: |
| + res.RetryAfter = parseRetryAfter(resp.Header.Get("Retry-After"), s.now()) |
| + res.Err = fmt.Errorf("aviso: push service throttled (%d)", resp.StatusCode) |
| + default: |
| + res.Err = fmt.Errorf("aviso: push service refused (%d)", resp.StatusCode) |
| + } |
| + return res |
| +} |
| + |
| +// redact strips the request URL from a transport error: *url.Error |
| +// prints it, and the endpoint is the one secret in this package that |
| +// would otherwise reach a log. What survives is the operation and the |
| +// underlying cause (a dial refusal, a timeout), which name at most the |
| +// push service's IP, never the endpoint's path. |
| +func redact(err error) error { |
| + var ue *url.Error |
| + if errors.As(err, &ue) { |
| + return fmt.Errorf("aviso: %s: %w", ue.Op, ue.Err) |
| + } |
| + return fmt.Errorf("aviso: send: %w", err) |
| +} |
| + |
| +func parseRetryAfter(v string, now time.Time) time.Duration { |
| + if v == "" { |
| + return 0 |
| + } |
| + if secs, err := strconv.Atoi(v); err == nil && secs >= 0 { |
| + return time.Duration(secs) * time.Second |
| + } |
| + if t, err := http.ParseTime(v); err == nil && t.After(now) { |
| + return t.Sub(now) |
| + } |
| + return 0 |
| +} |
diff --git a/send_test.go b/send_test.go| new file mode 100644 |
| index 0000000..54c80a5 |
| --- /dev/null |
| +++ b/send_test.go |
| @@ -0,0 +1,342 @@ |
| +package aviso |
| + |
| +import ( |
| + "context" |
| + "encoding/base64" |
| + "encoding/json" |
| + "errors" |
| + "net/http" |
| + "net/http/httptest" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// pushRecorder is a stand-in push service: it records each request's |
| +// headers and answers with whatever status the test sets. |
| +type pushRecorder struct { |
| + srv *httptest.Server |
| + status atomic.Int32 |
| + hdr chan http.Header |
| + retry string |
| + hold chan struct{} // when non-nil, handlers block until it closes |
| + inFlt atomic.Int32 |
| + peak atomic.Int32 |
| +} |
| + |
| +func newPushRecorder(t *testing.T) *pushRecorder { |
| + t.Helper() |
| + p := &pushRecorder{hdr: make(chan http.Header, 256)} |
| + p.status.Store(201) |
| + p.srv = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + n := p.inFlt.Add(1) |
| + for { |
| + old := p.peak.Load() |
| + if n <= old || p.peak.CompareAndSwap(old, n) { |
| + break |
| + } |
| + } |
| + defer p.inFlt.Add(-1) |
| + p.hdr <- r.Header.Clone() |
| + if p.hold != nil { |
| + <-p.hold |
| + } |
| + if p.retry != "" { |
| + w.Header().Set("Retry-After", p.retry) |
| + } |
| + w.WriteHeader(int(p.status.Load())) |
| + })) |
| + t.Cleanup(p.srv.Close) |
| + return p |
| +} |
| + |
| +// url is the recorder's address as an endpoint Send will accept: a |
| +// hostname, because validateEndpoint refuses a literal loopback IP |
| +// before any HTTP happens. Only the swapped-in client (below) lets the |
| +// name reach loopback at all. |
| +func (p *pushRecorder) url(path string) string { |
| + return strings.Replace(p.srv.URL, "127.0.0.1", "localhost", 1) + path |
| +} |
| + |
| +// serviceAgainst is the private seam the spec names: the SSRF guard |
| +// would refuse httptest's loopback server, so the test swaps in the |
| +// server's own client, pinned to the certificate's name so "localhost" |
| +// verifies. newClient's guards are tested on their own. |
| +func serviceAgainst(t *testing.T, p *pushRecorder) *Service { |
| + t.Helper() |
| + s := newInternalService(t) |
| + c := p.srv.Client() |
| + tr := c.Transport.(*http.Transport).Clone() |
| + tr.TLSClientConfig.ServerName = "example.com" |
| + c.Transport = tr |
| + s.client = c |
| + return s |
| +} |
| + |
| +func jwtSub(t *testing.T, authorization string) string { |
| + t.Helper() |
| + // "vapid t=<jwt>, k=<key>" |
| + i := strings.Index(authorization, "t=") |
| + j := strings.Index(authorization, ",") |
| + if i < 0 || j < i { |
| + t.Fatalf("authorization %q", authorization) |
| + } |
| + parts := strings.Split(authorization[i+2:j], ".") |
| + if len(parts) != 3 { |
| + t.Fatalf("jwt %q", authorization) |
| + } |
| + raw, err := base64.RawURLEncoding.DecodeString(parts[1]) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var claims struct{ Sub string } |
| + if err := json.Unmarshal(raw, &claims); err != nil { |
| + t.Fatal(err) |
| + } |
| + return claims.Sub |
| +} |
| + |
| +func TestSendToSetsHeadersAndConfirms(t *testing.T) { |
| + p := newPushRecorder(t) |
| + s := serviceAgainst(t, p) |
| + ctx := context.Background() |
| + _ = s.put(ctx, "alice", sub(p.url("/one")), "") |
| + s.now = func() time.Time { return time.Unix(1_800_000_000, 0) } |
| + res, err := s.SendTo(ctx, "alice", []byte(`{"title":"hi"}`), Options{TTL: 90 * time.Second, Urgency: "high", Topic: "t1"}) |
| + if err != nil || len(res) != 1 || res[0].Err != nil || res[0].Status != 201 { |
| + t.Fatalf("res=%+v err=%v", res, err) |
| + } |
| + h := <-p.hdr |
| + if h.Get("TTL") != "90" || h.Get("Urgency") != "high" || h.Get("Topic") != "t1" || |
| + h.Get("Content-Encoding") != "aes128gcm" { |
| + t.Fatalf("headers: %v", h) |
| + } |
| + if sub := jwtSub(t, h.Get("Authorization")); sub != "mailto:x@y" { |
| + t.Fatalf("VAPID sub = %q, want mailto:x@y (webpush-go adds the prefix itself)", sub) |
| + } |
| + var confirmed int64 |
| + _ = s.cfg.DB.QueryRow(`SELECT last_confirmed_at FROM aviso_subscriptions WHERE subject='alice'`).Scan(&confirmed) |
| + if confirmed != 1_800_000_000 { |
| + t.Fatalf("2xx did not confirm: %d", confirmed) |
| + } |
| +} |
| + |
| +func TestSendDefaultsTTLToADayAndUrgencyToNormal(t *testing.T) { |
| + p := newPushRecorder(t) |
| + s := serviceAgainst(t, p) |
| + ctx := context.Background() |
| + _ = s.put(ctx, "alice", sub(p.url("/one")), "") |
| + if _, err := s.SendTo(ctx, "alice", []byte("x"), Options{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + h := <-p.hdr |
| + if h.Get("TTL") != "86400" || h.Get("Urgency") != "normal" || h.Get("Topic") != "" { |
| + t.Fatalf("headers: TTL=%q Urgency=%q Topic=%q", h.Get("TTL"), h.Get("Urgency"), h.Get("Topic")) |
| + } |
| +} |
| + |
| +func TestSendPrunesOnGoneOnlyAtSameRevision(t *testing.T) { |
| + p := newPushRecorder(t) |
| + s := serviceAgainst(t, p) |
| + ctx := context.Background() |
| + _ = s.put(ctx, "alice", sub(p.url("/one")), "") |
| + rows, _ := s.List(ctx, "alice") |
| + stale := rows[0] |
| + _ = s.put(ctx, "alice", sub(p.url("/one")), "") // revision 2 |
| + p.status.Store(410) |
| + res, _ := s.Send(ctx, []Stored{stale}, []byte("x"), Options{}) |
| + if res[0].Status != 410 || res[0].Err == nil { |
| + t.Fatalf("res=%+v", res) |
| + } |
| + if rows, _ := s.List(ctx, "alice"); len(rows) != 1 { |
| + t.Fatal("410 at a stale revision pruned a refreshed row") |
| + } |
| + current, _ := s.List(ctx, "alice") |
| + _, _ = s.Send(ctx, current, []byte("x"), Options{}) |
| + if rows, _ := s.List(ctx, "alice"); len(rows) != 0 { |
| + t.Fatal("410 at the current revision did not prune") |
| + } |
| +} |
| + |
| +func TestSendReportsRetryAfter(t *testing.T) { |
| + p := newPushRecorder(t) |
| + p.retry = "120" |
| + p.status.Store(429) |
| + s := serviceAgainst(t, p) |
| + ctx := context.Background() |
| + _ = s.put(ctx, "alice", sub(p.url("/one")), "") |
| + res, _ := s.SendTo(ctx, "alice", []byte("x"), Options{}) |
| + if res[0].RetryAfter != 120*time.Second || res[0].Err == nil || res[0].Status != 429 { |
| + t.Fatalf("res=%+v", res) |
| + } |
| + if rows, _ := s.List(ctx, "alice"); len(rows) != 1 { |
| + t.Fatal("429 pruned") |
| + } |
| +} |
| + |
| +func TestSendSkipsRowsUnderAnotherKey(t *testing.T) { |
| + p := newPushRecorder(t) |
| + s := serviceAgainst(t, p) |
| + ctx := context.Background() |
| + _ = s.put(ctx, "alice", sub(p.url("/one")), "") |
| + _, _ = s.cfg.DB.Exec(`UPDATE aviso_subscriptions SET vapid_key_id = 'other'`) |
| + res, err := s.SendTo(ctx, "alice", []byte("x"), Options{}) |
| + if err != nil || len(res) != 1 || !errors.Is(res[0].Err, ErrKeyMismatch) { |
| + t.Fatalf("res=%+v err=%v", res, err) |
| + } |
| + select { |
| + case <-p.hdr: |
| + t.Fatal("sent despite key mismatch") |
| + default: |
| + } |
| +} |
| + |
| +func TestSendRefusesAnEditedEndpoint(t *testing.T) { |
| + p := newPushRecorder(t) |
| + s := serviceAgainst(t, p) |
| + ctx := context.Background() |
| + _ = s.put(ctx, "alice", sub(p.url("/one")), "") |
| + rows, _ := s.List(ctx, "alice") |
| + rows[0].Endpoint = strings.Replace(rows[0].Endpoint, "https://", "http://", 1) |
| + res, err := s.Send(ctx, rows, []byte("x"), Options{}) |
| + if err != nil || !errors.Is(res[0].Err, ErrBadEndpoint) { |
| + t.Fatalf("res=%+v err=%v", res, err) |
| + } |
| + select { |
| + case <-p.hdr: |
| + t.Fatal("sent to an http endpoint") |
| + default: |
| + } |
| +} |
| + |
| +func TestSendValidatesPayloadAndOptions(t *testing.T) { |
| + s := newInternalService(t) |
| + ctx := context.Background() |
| + if _, err := s.Send(ctx, nil, make([]byte, 3994), Options{}); !errors.Is(err, ErrPayloadTooLarge) { |
| + t.Errorf("oversize payload: %v", err) |
| + } |
| + if _, err := s.Send(ctx, nil, make([]byte, 3993), Options{}); err != nil { |
| + t.Errorf("3993 bytes refused: %v", err) |
| + } |
| + for name, o := range map[string]Options{ |
| + "neg ttl": {TTL: -time.Second}, |
| + "frac ttl": {TTL: 1500 * time.Millisecond}, |
| + "urgency": {Urgency: "urgent"}, |
| + "topic chars": {Topic: "a b"}, |
| + "topic long": {Topic: strings.Repeat("a", 33)}, |
| + } { |
| + if _, err := s.Send(ctx, nil, []byte("x"), o); !errors.Is(err, ErrBadOptions) { |
| + t.Errorf("%s: %v", name, err) |
| + } |
| + } |
| +} |
| + |
| +func TestSendHonoursCancellation(t *testing.T) { |
| + p := newPushRecorder(t) |
| + s := serviceAgainst(t, p) |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + cancel() |
| + _ = s.put(context.Background(), "alice", sub(p.url("/one")), "") |
| + _, err := s.SendTo(ctx, "alice", []byte("x"), Options{}) |
| + if !errors.Is(err, context.Canceled) { |
| + t.Fatalf("got %v, want context.Canceled", err) |
| + } |
| +} |
| + |
| +// Cancellation while sends are in flight: rows never started carry the |
| +// context error, started ones finish, the batch error is the context's, |
| +// and Send returns only after every goroutine it started has stopped |
| +// writing — which is what -race checks here. |
| +func TestSendCancelledMidBatchWaitsForStartedGoroutines(t *testing.T) { |
| + p := newPushRecorder(t) |
| + p.hold = make(chan struct{}) |
| + s := serviceAgainst(t, p) |
| + s.sem = make(chan struct{}, 2) // room for two in flight, the rest must wait |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + var rows []Stored |
| + for i := 0; i < 5; i++ { |
| + _ = s.put(context.Background(), "alice", sub(p.url("/"+string(rune('a'+i)))), "") |
| + } |
| + rows, _ = s.List(context.Background(), "alice") |
| + var wg sync.WaitGroup |
| + wg.Add(1) |
| + var res []Result |
| + var err error |
| + go func() { |
| + defer wg.Done() |
| + res, err = s.Send(ctx, rows, []byte("x"), Options{}) |
| + }() |
| + <-p.hdr |
| + <-p.hdr // two are in flight and blocked |
| + cancel() |
| + close(p.hold) |
| + wg.Wait() |
| + if !errors.Is(err, context.Canceled) { |
| + t.Fatalf("batch err = %v", err) |
| + } |
| + // A row never started carries the bare context error; a row whose |
| + // request was cut short carries it wrapped by the transport (and |
| + // was seen by the recorder). |
| + started, skipped := 0, 0 |
| + for _, r := range res { |
| + switch { |
| + case r.Err == context.Canceled: |
| + skipped++ |
| + case r.Err != nil: |
| + started++ |
| + default: |
| + t.Fatalf("result with no error after cancellation: %+v", r) |
| + } |
| + } |
| + if started != 2 || skipped != 3 { |
| + t.Fatalf("started=%d skipped=%d, want 2/3", started, skipped) |
| + } |
| + if seen := len(p.hdr) + 2; seen != 2 { // the two we drained; nothing else reached the service |
| + t.Fatalf("recorder saw %d requests, want 2", seen) |
| + } |
| +} |
| + |
| +func TestSendBoundsConcurrency(t *testing.T) { |
| + p := newPushRecorder(t) |
| + p.hold = make(chan struct{}) |
| + s := serviceAgainst(t, p) |
| + s.sem = make(chan struct{}, 3) |
| + ctx := context.Background() |
| + for i := 0; i < 8; i++ { |
| + _ = s.put(ctx, "alice", sub(p.url("/"+string(rune('a'+i)))), "") |
| + } |
| + done := make(chan struct{}) |
| + go func() { _, _ = s.SendTo(ctx, "alice", []byte("x"), Options{}); close(done) }() |
| + for i := 0; i < 3; i++ { |
| + <-p.hdr |
| + } |
| + time.Sleep(50 * time.Millisecond) // give a fourth a chance to (wrongly) start |
| + if got := p.inFlt.Load(); got != 3 { |
| + t.Fatalf("in flight = %d, want 3", got) |
| + } |
| + close(p.hold) |
| + <-done |
| + if p.peak.Load() > 3 { |
| + t.Fatalf("peak concurrency %d exceeded bound 3", p.peak.Load()) |
| + } |
| +} |
| + |
| +func TestSendRedactsEndpointFromErrors(t *testing.T) { |
| + s := newInternalService(t) |
| + ctx := context.Background() |
| + // Routable-looking host, refused at dial by the guard once it |
| + // resolves — or unreachable; either way the error must not carry |
| + // the URL's path, which is the secret part of an endpoint. |
| + _ = s.put(ctx, "alice", sub("https://127.0.0.1:9/secret-path"), "") |
| + rows, _ := s.List(ctx, "alice") |
| + rows[0].Endpoint = "https://localhost:9/secret-path" // literal IP would fail validation; a name reaches the dialer |
| + res, _ := s.Send(ctx, rows, []byte("x"), Options{}) |
| + if res[0].Err == nil || strings.Contains(res[0].Err.Error(), "secret-path") { |
| + t.Fatalf("error carries the endpoint: %v", res[0].Err) |
| + } |
| + if !strings.Contains(res[0].Err.Error(), "aviso") { |
| + t.Fatalf("error not package-prefixed: %v", res[0].Err) |
| + } |
| +} |