mirror of
https://github.com/acaloiaro/newsbox
synced 2026-07-21 02:09:55 +00:00
Refactor to use neoq
This commit is contained in:
Adriano Caloiaro
parent
8ae801ae5f
commit
586e18defb
12 changed files with 215 additions and 390 deletions
22
.buffalo.dev.yml
Normal file
22
.buffalo.dev.yml
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
app_root: .
|
||||
ignored_folders:
|
||||
- vendor
|
||||
- log
|
||||
- logs
|
||||
- assets
|
||||
- public
|
||||
- grifts
|
||||
- tmp
|
||||
- bin
|
||||
- node_modules
|
||||
- .sass-cache
|
||||
included_extensions:
|
||||
- .go
|
||||
- .env
|
||||
build_path: tmp
|
||||
build_delay: 200ns
|
||||
build_target_path: "./cmd/app"
|
||||
binary_name: foobar-build
|
||||
command_flags: []
|
||||
enable_colors: true
|
||||
log_name: buffalo
|
||||
|
|
@ -8,6 +8,7 @@ import (
|
|||
"newsbox/locales"
|
||||
"newsbox/models"
|
||||
"newsbox/public"
|
||||
"newsbox/workers"
|
||||
|
||||
"github.com/gobuffalo/buffalo"
|
||||
"github.com/gobuffalo/buffalo-pop/v3/pop/popmw"
|
||||
|
|
@ -31,6 +32,7 @@ var (
|
|||
|
||||
func App() *buffalo.App {
|
||||
if app == nil {
|
||||
workers.Start()
|
||||
app = buffalo.New(buffalo.Options{
|
||||
Env: ENV,
|
||||
SessionName: "_newsbox_session",
|
||||
|
|
|
|||
|
|
@ -8,9 +8,10 @@ import (
|
|||
"net/http"
|
||||
"newsbox/internal"
|
||||
"newsbox/models"
|
||||
"newsbox/workers"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
"github.com/gobuffalo/buffalo"
|
||||
"github.com/gobuffalo/envy"
|
||||
"github.com/gobuffalo/pop/v6"
|
||||
|
|
@ -20,6 +21,11 @@ import (
|
|||
|
||||
// MailgunWebhook handles new message webhooks from Mailgun
|
||||
func MailgunWebhook(c buffalo.Context) (err error) {
|
||||
dbURL, err := envy.MustGet("DATABASE_URL")
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to get DATABASE_URL")
|
||||
}
|
||||
|
||||
mgApiKey, err := envy.MustGet("MAILGUN_API_KEY")
|
||||
if err != nil {
|
||||
log.Println("MAILGUN_API_KEY not set")
|
||||
|
|
@ -79,10 +85,14 @@ func MailgunWebhook(c buffalo.Context) (err error) {
|
|||
}
|
||||
|
||||
js, _ := json.MarshalIndent(&m, "", "\t")
|
||||
_, err = workers.W.EnqueueJob("incoming_email", js)
|
||||
nq, _ := neoq.New(dbURL)
|
||||
_, err = nq.Enqueue(neoq.Job{
|
||||
Queue: "incoming_email",
|
||||
Payload: map[string]interface{}{
|
||||
"message_id": m.ID,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
c.Response().Write([]byte("Unable to queue message"))
|
||||
c.Response().WriteHeader(400)
|
||||
return
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,11 +1,11 @@
|
|||
package actions
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
"github.com/gobuffalo/buffalo"
|
||||
"github.com/gobuffalo/pop/v6"
|
||||
"github.com/gobuffalo/x/responder"
|
||||
|
|
@ -13,9 +13,7 @@ import (
|
|||
"github.com/pkg/errors"
|
||||
|
||||
"newsbox/internal"
|
||||
"newsbox/mill"
|
||||
"newsbox/models"
|
||||
"newsbox/workers"
|
||||
)
|
||||
|
||||
// Show gets the profile data for a user and renders it
|
||||
|
|
@ -155,13 +153,16 @@ func UpdateProfile(c buffalo.Context) error {
|
|||
}
|
||||
|
||||
invitationAddr := user.Email.Interface().(string)
|
||||
params := mill.EmailAddressChangeEmailJob{
|
||||
Recipient: invitationAddr,
|
||||
VerificationURL: fmt.Sprintf("%s/users/email_verification/%s", internal.BaseUrl(), uev.ID),
|
||||
emailAddrChangedJob := neoq.Job{
|
||||
Queue: "email_verifications",
|
||||
Payload: map[string]any{
|
||||
"recipient": invitationAddr,
|
||||
"verification_url": fmt.Sprintf("%s/users/email_verification/%s", internal.BaseUrl(), uev.ID),
|
||||
},
|
||||
}
|
||||
paramsJSON, _ := json.Marshal(params)
|
||||
|
||||
_, err = workers.W.EnqueueJob("email_address_changes", paramsJSON)
|
||||
nq, _ := neoq.New(internal.DatabaseUrl())
|
||||
_, err = nq.Enqueue(emailAddrChangedJob)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,17 +1,15 @@
|
|||
package actions
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
"github.com/gobuffalo/buffalo"
|
||||
"github.com/gobuffalo/pop/v6"
|
||||
"github.com/pkg/errors"
|
||||
|
||||
"newsbox/internal"
|
||||
"newsbox/mill"
|
||||
"newsbox/models"
|
||||
"newsbox/workers"
|
||||
)
|
||||
|
||||
// UsersNew renders the users form
|
||||
|
|
@ -47,13 +45,16 @@ func UsersCreate(c buffalo.Context) error {
|
|||
}
|
||||
|
||||
invitationAddr := u.Email.Interface().(string)
|
||||
params := mill.WelcomeEmailJob{
|
||||
Recipient: invitationAddr,
|
||||
VerificationURL: fmt.Sprintf("%s/users/email_verification/%s", internal.BaseUrl(), uev.ID),
|
||||
welcomeEmailJob := neoq.Job{
|
||||
Queue: "welcome_emails",
|
||||
Payload: map[string]any{
|
||||
"recipient": invitationAddr,
|
||||
"verification_url": fmt.Sprintf("%s/users/email_verification/%s", internal.BaseUrl(), uev.ID),
|
||||
},
|
||||
}
|
||||
paramsJSON, _ := json.Marshal(params)
|
||||
|
||||
_, err = workers.W.EnqueueJob("welcome_emails", paramsJSON)
|
||||
nq, _ := neoq.New(internal.DatabaseUrl())
|
||||
_, err = nq.Enqueue(welcomeEmailJob)
|
||||
if err != nil {
|
||||
c.Redirect(302, "/")
|
||||
return err
|
||||
|
|
|
|||
15
go.mod
15
go.mod
|
|
@ -2,9 +2,10 @@ module newsbox
|
|||
|
||||
go 1.18
|
||||
|
||||
replace github.com/acaloiaro/neoq v0.0.0-20230215002938-fc011c9c8409 => /home/adriano/git/neoq
|
||||
|
||||
require (
|
||||
github.com/btubbs/pgq v0.0.0-20190101193147-0a3335913e86
|
||||
github.com/gin-gonic/gin v1.8.2
|
||||
github.com/acaloiaro/neoq v0.0.0-20230215002938-fc011c9c8409
|
||||
github.com/gobuffalo/buffalo v0.18.9
|
||||
github.com/gobuffalo/buffalo-pop/v3 v3.0.6
|
||||
github.com/gobuffalo/envy v1.10.2
|
||||
|
|
@ -19,7 +20,6 @@ require (
|
|||
github.com/gobuffalo/validate/v3 v3.3.3
|
||||
github.com/gobuffalo/x v0.1.0
|
||||
github.com/gofrs/uuid v4.3.0+incompatible
|
||||
github.com/jackc/pgx/v5 v5.2.0
|
||||
github.com/mailgun/mailgun-go/v4 v4.8.1
|
||||
github.com/markbates/going v1.0.3
|
||||
github.com/markbates/goth v1.76.0
|
||||
|
|
@ -53,7 +53,6 @@ require (
|
|||
github.com/gobuffalo/plush/v4 v4.1.16 // indirect
|
||||
github.com/gobuffalo/refresh v1.13.2 // indirect
|
||||
github.com/gobuffalo/tags/v3 v3.1.4 // indirect
|
||||
github.com/goccy/go-json v0.9.11 // indirect
|
||||
github.com/golang/protobuf v1.5.0 // indirect
|
||||
github.com/gorilla/css v1.0.0 // indirect
|
||||
github.com/gorilla/handlers v1.5.1 // indirect
|
||||
|
|
@ -63,17 +62,17 @@ require (
|
|||
github.com/guregu/null v4.0.0+incompatible // indirect
|
||||
github.com/inconshreveable/mousetrap v1.0.0 // indirect
|
||||
github.com/jackc/chunkreader/v2 v2.0.1 // indirect
|
||||
github.com/jackc/pgconn v1.12.1 // indirect
|
||||
github.com/jackc/pgconn v1.13.0 // indirect
|
||||
github.com/jackc/pgio v1.0.0 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgproto3/v2 v2.3.0 // indirect
|
||||
github.com/jackc/pgproto3/v2 v2.3.1 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
|
||||
github.com/jackc/pgtype v1.11.0 // indirect
|
||||
github.com/jackc/pgx/v4 v4.16.1 // indirect
|
||||
github.com/jackc/pgx/v5 v5.2.0 // indirect
|
||||
github.com/jackc/puddle/v2 v2.1.2 // indirect
|
||||
github.com/jmoiron/sqlx v1.3.5 // indirect
|
||||
github.com/joho/godotenv v1.4.0 // indirect
|
||||
github.com/joomcode/errorx v1.1.0 // indirect
|
||||
github.com/json-iterator/go v1.1.12 // indirect
|
||||
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 // indirect
|
||||
github.com/kr/text v0.2.0 // indirect
|
||||
|
|
@ -89,7 +88,6 @@ require (
|
|||
github.com/monoculum/formam v3.5.5+incompatible // indirect
|
||||
github.com/nicksnyder/go-i18n v1.10.1 // indirect
|
||||
github.com/pelletier/go-toml v1.2.0 // indirect
|
||||
github.com/pelletier/go-toml/v2 v2.0.6 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
github.com/robfig/cron v1.2.0 // indirect
|
||||
github.com/rogpeppe/go-internal v1.9.0 // indirect
|
||||
|
|
@ -100,7 +98,6 @@ require (
|
|||
github.com/spf13/cobra v1.5.0 // indirect
|
||||
github.com/spf13/pflag v1.0.5 // indirect
|
||||
github.com/stretchr/testify v1.8.1 // indirect
|
||||
github.com/ugorji/go/codec v1.2.7 // indirect
|
||||
go.uber.org/atomic v1.10.0 // indirect
|
||||
golang.org/x/net v0.4.0 // indirect
|
||||
golang.org/x/oauth2 v0.0.0-20200902213428-5d25da1a8d43 // indirect
|
||||
|
|
|
|||
20
go.sum
20
go.sum
|
|
@ -43,8 +43,6 @@ github.com/acaloiaro/buffalo v1.1.1 h1:/Oj6NhhlogpnGfsOMhGjpDNthPosXN42PZ/W3/yCF
|
|||
github.com/acaloiaro/buffalo v1.1.1/go.mod h1:n3GIXIy7fZWTgn09AZUHZc8pc/1UGArODsdN4xIRjkY=
|
||||
github.com/aymerick/douceur v0.2.0 h1:Mv+mAeH1Q+n9Fr+oyamOlAkUNPWPlA8PPGR0QAaYuPk=
|
||||
github.com/aymerick/douceur v0.2.0/go.mod h1:wlT5vV2O3h55X9m7iVYN0TBM0NH/MmbLnd30/FjWUq4=
|
||||
github.com/btubbs/pgq v0.0.0-20190101193147-0a3335913e86 h1:VWFMCbz4J10pLBcNN/GCjiGvmcPrBsNaa8XCvjhnG0s=
|
||||
github.com/btubbs/pgq v0.0.0-20190101193147-0a3335913e86/go.mod h1:/zonZGp1GzMn8IFsphjvGCyHNYWXxV7akoX6IJV3qEo=
|
||||
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
|
||||
github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
|
||||
github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
|
||||
|
|
@ -84,8 +82,6 @@ github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSw
|
|||
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
|
||||
github.com/fsnotify/fsnotify v1.5.4 h1:jRbGcIw6P2Meqdwuo0H1p6JVLbL5DHKAKlYndzMwVZI=
|
||||
github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=
|
||||
github.com/gin-gonic/gin v1.8.2 h1:UzKToD9/PoFj/V4rvlKqTRKnQYyz8Sc1MJlv4JHPtvY=
|
||||
github.com/gin-gonic/gin v1.8.2/go.mod h1:qw5AYuDrzRTnhvusDsrov+fDIxp9Dleuu12h8nfB398=
|
||||
github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
|
||||
|
|
@ -157,8 +153,6 @@ github.com/gobuffalo/validate/v3 v3.3.3/go.mod h1:YC7FsbJ/9hW/VjQdmXPvFqvRis4vrR
|
|||
github.com/gobuffalo/x v0.1.0 h1:ILV6PIfyQto7RKfxRutQUuW234x+A5/+FRQpik0hNrM=
|
||||
github.com/gobuffalo/x v0.1.0/go.mod h1:WevpGD+5YOreDJznWevcn8NTmQEW5STSBgIkpkjzqXc=
|
||||
github.com/goccy/go-json v0.9.6/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
|
||||
github.com/goccy/go-json v0.9.11 h1:/pAaQDLHEoCq/5FFmSKBswWmK6H0e8g4159Kc/X/nqk=
|
||||
github.com/goccy/go-json v0.9.11/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
|
||||
github.com/gofrs/uuid v4.0.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
|
||||
github.com/gofrs/uuid v4.1.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
|
||||
github.com/gofrs/uuid v4.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
|
||||
|
|
@ -250,8 +244,9 @@ 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 h1:rsDFzIpRk7xT4B8FufgpCCeyjdNpKyghZeSefViE5W8=
|
||||
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=
|
||||
github.com/jackc/pgio v1.0.0/go.mod h1:oP+2QK2wFfUWgr+gxjoBH9KGBb31Eio69xUb0w5bYf8=
|
||||
github.com/jackc/pgmock v0.0.0-20190831213851-13a1b77aafa2/go.mod h1:fGZlG77KXmcq05nJLRkk0+p82V8B8Dw8KN2/V9c/OAE=
|
||||
|
|
@ -267,8 +262,9 @@ 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 h1:brH0pCGBDkBW07HWlN/oSBXrmo3WB0UvZd1pIuDcL8Y=
|
||||
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=
|
||||
github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b/go.mod h1:vsD4gTJCa9TptPL8sPkXrLZ+hDuNrZCnj29CQpr4X1E=
|
||||
github.com/jackc/pgtype v0.0.0-20190421001408-4ed0de4755e0/go.mod h1:hdSHsc1V01CGwFsrv11mJRHWJ6aifDLfdV3aVjFF0zg=
|
||||
|
|
@ -297,8 +293,6 @@ github.com/jmoiron/sqlx v1.3.5 h1:vFFPA71p1o5gAeqtEAwLU4dnX2napprKtHr7PYIcN3g=
|
|||
github.com/jmoiron/sqlx v1.3.5/go.mod h1:nRVWtLre0KfCLJvgxzCsLVMogSvQ1zNJtpYr2Ccp0mQ=
|
||||
github.com/joho/godotenv v1.4.0 h1:3l4+N6zfMWnkbPEXKng2o2/MR5mSwTrBih4ZEkkz1lg=
|
||||
github.com/joho/godotenv v1.4.0/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
|
||||
github.com/joomcode/errorx v1.1.0 h1:dizuSG6yHzlvXOOGHW00gwsmM4Sb9x/yWEfdtPztqcs=
|
||||
github.com/joomcode/errorx v1.1.0/go.mod h1:eQzdtdlNyN7etw6YCS4W4+lu442waxZYw5yvz0ULrRo=
|
||||
github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
|
||||
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
|
||||
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
|
||||
|
|
@ -374,8 +368,6 @@ github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+W
|
|||
github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
|
||||
github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
|
||||
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
|
||||
github.com/pelletier/go-toml/v2 v2.0.6 h1:nrzqCb7j9cDFj2coyLNLaZuJTLjWjlaz6nvTvIwycIU=
|
||||
github.com/pelletier/go-toml/v2 v2.0.6/go.mod h1:eumQOmlWiOPt5WriQQqoM5y18pDHwha2N+QD+EUNTek=
|
||||
github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
|
||||
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
|
|
@ -430,9 +422,6 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
|
|||
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
|
||||
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
|
||||
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
|
||||
github.com/ugorji/go v1.2.7/go.mod h1:nF9osbDWLy6bDVv/Rtoh6QgnvNDpmCalQV5urGCCS6M=
|
||||
github.com/ugorji/go/codec v1.2.7 h1:YPXUKf7fYbp/y8xloBqZOw2qaVggbfwMlI8WM3wZUJ0=
|
||||
github.com/ugorji/go/codec v1.2.7/go.mod h1:WGN1fab3R1fzQlVQTkfxVtIBhWDRqOviHU95kRgeqEY=
|
||||
github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
|
|
@ -470,6 +459,7 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
|
|||
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
|
||||
golang.org/x/crypto v0.0.0-20220214200702-86341886e292/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
|
||||
golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
|
||||
|
|
|
|||
|
|
@ -1,11 +1,23 @@
|
|||
package internal
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
|
||||
"github.com/gobuffalo/envy"
|
||||
)
|
||||
|
||||
// DatabaseURL returns the site's database URL
|
||||
func DatabaseUrl() string {
|
||||
dbURL, err := envy.MustGet("DATABASE_URL")
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to get DATABASE_URL")
|
||||
}
|
||||
|
||||
return dbURL
|
||||
}
|
||||
|
||||
// SiteName returns the name of the current site
|
||||
func SiteName() string {
|
||||
return envy.Get("SITE_NAME", "Newsbox")
|
||||
|
|
|
|||
|
|
@ -21,5 +21,6 @@ func SendWelcomeEmail(recipient, verificationUrl string) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return smtp.Send(m)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,66 +16,12 @@ SET xmloption = content;
|
|||
SET client_min_messages = warning;
|
||||
SET row_security = off;
|
||||
|
||||
--
|
||||
-- Name: job_status; Type: TYPE; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
CREATE TYPE public.job_status AS ENUM (
|
||||
'new',
|
||||
'running',
|
||||
'processed',
|
||||
'failed'
|
||||
);
|
||||
|
||||
|
||||
ALTER TYPE public.job_status OWNER TO newsbox;
|
||||
|
||||
SET default_tablespace = '';
|
||||
|
||||
SET default_table_access_method = heap;
|
||||
|
||||
--
|
||||
-- Name: jobs; Type: TABLE; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
CREATE TABLE public.jobs (
|
||||
id integer NOT NULL,
|
||||
type text NOT NULL,
|
||||
queue text NOT NULL,
|
||||
payload jsonb NOT NULL,
|
||||
status public.job_status NOT NULL,
|
||||
retries integer DEFAULT 0 NOT NULL,
|
||||
created_at timestamp without time zone DEFAULT now(),
|
||||
updated_at timestamp without time zone
|
||||
);
|
||||
|
||||
|
||||
ALTER TABLE public.jobs OWNER TO newsbox;
|
||||
|
||||
--
|
||||
-- Name: jobs_id_seq; Type: SEQUENCE; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
CREATE SEQUENCE public.jobs_id_seq
|
||||
AS integer
|
||||
START WITH 1
|
||||
INCREMENT BY 1
|
||||
NO MINVALUE
|
||||
NO MAXVALUE
|
||||
CACHE 1;
|
||||
|
||||
|
||||
ALTER TABLE public.jobs_id_seq OWNER TO newsbox;
|
||||
|
||||
--
|
||||
-- Name: jobs_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
ALTER SEQUENCE public.jobs_id_seq OWNED BY public.jobs.id;
|
||||
|
||||
|
||||
--
|
||||
-- Name: message_deliveries; Type: TABLE; Schema: public; Owner: newsbox
|
||||
-- Name: message_deliveries; Type: TABLE; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
CREATE TABLE public.message_deliveries (
|
||||
|
|
@ -86,10 +32,10 @@ CREATE TABLE public.message_deliveries (
|
|||
);
|
||||
|
||||
|
||||
ALTER TABLE public.message_deliveries OWNER TO newsbox;
|
||||
ALTER TABLE public.message_deliveries OWNER TO postgres;
|
||||
|
||||
--
|
||||
-- Name: messages; Type: TABLE; Schema: public; Owner: newsbox
|
||||
-- Name: messages; Type: TABLE; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
CREATE TABLE public.messages (
|
||||
|
|
@ -107,50 +53,10 @@ CREATE TABLE public.messages (
|
|||
);
|
||||
|
||||
|
||||
ALTER TABLE public.messages OWNER TO newsbox;
|
||||
ALTER TABLE public.messages OWNER TO postgres;
|
||||
|
||||
--
|
||||
-- Name: pgq_jobs; Type: TABLE; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
CREATE TABLE public.pgq_jobs (
|
||||
id integer NOT NULL,
|
||||
created_at timestamp with time zone DEFAULT now() NOT NULL,
|
||||
queue_name text NOT NULL,
|
||||
data bytea NOT NULL,
|
||||
run_after timestamp with time zone NOT NULL,
|
||||
retry_waits text[] NOT NULL,
|
||||
ran_at timestamp with time zone,
|
||||
error text
|
||||
);
|
||||
|
||||
|
||||
ALTER TABLE public.pgq_jobs OWNER TO newsbox;
|
||||
|
||||
--
|
||||
-- Name: pgq_jobs_id_seq; Type: SEQUENCE; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
CREATE SEQUENCE public.pgq_jobs_id_seq
|
||||
AS integer
|
||||
START WITH 1
|
||||
INCREMENT BY 1
|
||||
NO MINVALUE
|
||||
NO MAXVALUE
|
||||
CACHE 1;
|
||||
|
||||
|
||||
ALTER TABLE public.pgq_jobs_id_seq OWNER TO newsbox;
|
||||
|
||||
--
|
||||
-- Name: pgq_jobs_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
ALTER SEQUENCE public.pgq_jobs_id_seq OWNED BY public.pgq_jobs.id;
|
||||
|
||||
|
||||
--
|
||||
-- Name: schema_migration; Type: TABLE; Schema: public; Owner: newsbox
|
||||
-- Name: schema_migration; Type: TABLE; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
CREATE TABLE public.schema_migration (
|
||||
|
|
@ -158,10 +64,10 @@ CREATE TABLE public.schema_migration (
|
|||
);
|
||||
|
||||
|
||||
ALTER TABLE public.schema_migration OWNER TO newsbox;
|
||||
ALTER TABLE public.schema_migration OWNER TO postgres;
|
||||
|
||||
--
|
||||
-- Name: user_email_verifications; Type: TABLE; Schema: public; Owner: newsbox
|
||||
-- Name: user_email_verifications; Type: TABLE; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
CREATE TABLE public.user_email_verifications (
|
||||
|
|
@ -172,10 +78,10 @@ CREATE TABLE public.user_email_verifications (
|
|||
);
|
||||
|
||||
|
||||
ALTER TABLE public.user_email_verifications OWNER TO newsbox;
|
||||
ALTER TABLE public.user_email_verifications OWNER TO postgres;
|
||||
|
||||
--
|
||||
-- Name: user_preferences; Type: TABLE; Schema: public; Owner: newsbox
|
||||
-- Name: user_preferences; Type: TABLE; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
CREATE TABLE public.user_preferences (
|
||||
|
|
@ -188,10 +94,10 @@ CREATE TABLE public.user_preferences (
|
|||
);
|
||||
|
||||
|
||||
ALTER TABLE public.user_preferences OWNER TO newsbox;
|
||||
ALTER TABLE public.user_preferences OWNER TO postgres;
|
||||
|
||||
--
|
||||
-- Name: users; Type: TABLE; Schema: public; Owner: newsbox
|
||||
-- Name: users; Type: TABLE; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
CREATE TABLE public.users (
|
||||
|
|
@ -207,32 +113,10 @@ CREATE TABLE public.users (
|
|||
);
|
||||
|
||||
|
||||
ALTER TABLE public.users OWNER TO newsbox;
|
||||
ALTER TABLE public.users OWNER TO postgres;
|
||||
|
||||
--
|
||||
-- Name: jobs id; Type: DEFAULT; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
ALTER TABLE ONLY public.jobs ALTER COLUMN id SET DEFAULT nextval('public.jobs_id_seq'::regclass);
|
||||
|
||||
|
||||
--
|
||||
-- Name: pgq_jobs id; Type: DEFAULT; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
ALTER TABLE ONLY public.pgq_jobs ALTER COLUMN id SET DEFAULT nextval('public.pgq_jobs_id_seq'::regclass);
|
||||
|
||||
|
||||
--
|
||||
-- Name: jobs jobs_pkey; Type: CONSTRAINT; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
ALTER TABLE ONLY public.jobs
|
||||
ADD CONSTRAINT jobs_pkey PRIMARY KEY (id);
|
||||
|
||||
|
||||
--
|
||||
-- Name: message_deliveries message_deliveries_pkey; Type: CONSTRAINT; Schema: public; Owner: newsbox
|
||||
-- Name: message_deliveries message_deliveries_pkey; Type: CONSTRAINT; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
ALTER TABLE ONLY public.message_deliveries
|
||||
|
|
@ -240,7 +124,7 @@ ALTER TABLE ONLY public.message_deliveries
|
|||
|
||||
|
||||
--
|
||||
-- Name: messages messages_pkey; Type: CONSTRAINT; Schema: public; Owner: newsbox
|
||||
-- Name: messages messages_pkey; Type: CONSTRAINT; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
ALTER TABLE ONLY public.messages
|
||||
|
|
@ -248,15 +132,7 @@ ALTER TABLE ONLY public.messages
|
|||
|
||||
|
||||
--
|
||||
-- Name: pgq_jobs pgq_jobs_pkey; Type: CONSTRAINT; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
ALTER TABLE ONLY public.pgq_jobs
|
||||
ADD CONSTRAINT pgq_jobs_pkey PRIMARY KEY (id);
|
||||
|
||||
|
||||
--
|
||||
-- Name: user_email_verifications user_email_verifications_pkey; Type: CONSTRAINT; Schema: public; Owner: newsbox
|
||||
-- Name: user_email_verifications user_email_verifications_pkey; Type: CONSTRAINT; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
ALTER TABLE ONLY public.user_email_verifications
|
||||
|
|
@ -264,7 +140,7 @@ ALTER TABLE ONLY public.user_email_verifications
|
|||
|
||||
|
||||
--
|
||||
-- Name: user_preferences user_preferences_pkey; Type: CONSTRAINT; Schema: public; Owner: newsbox
|
||||
-- Name: user_preferences user_preferences_pkey; Type: CONSTRAINT; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
ALTER TABLE ONLY public.user_preferences
|
||||
|
|
@ -272,7 +148,7 @@ ALTER TABLE ONLY public.user_preferences
|
|||
|
||||
|
||||
--
|
||||
-- Name: users users_pkey; Type: CONSTRAINT; Schema: public; Owner: newsbox
|
||||
-- Name: users users_pkey; Type: CONSTRAINT; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
ALTER TABLE ONLY public.users
|
||||
|
|
@ -280,42 +156,21 @@ ALTER TABLE ONLY public.users
|
|||
|
||||
|
||||
--
|
||||
-- Name: idx_pgq_jobs_fetch; Type: INDEX; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
CREATE INDEX idx_pgq_jobs_fetch ON public.pgq_jobs USING btree (queue_name, run_after) WHERE (ran_at IS NULL);
|
||||
|
||||
|
||||
--
|
||||
-- Name: jobs_created_at_idx; Type: INDEX; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
CREATE INDEX jobs_created_at_idx ON public.jobs USING btree (created_at);
|
||||
|
||||
|
||||
--
|
||||
-- Name: jobs_queue_status_idx; Type: INDEX; Schema: public; Owner: newsbox
|
||||
--
|
||||
|
||||
CREATE INDEX jobs_queue_status_idx ON public.jobs USING btree (queue, status);
|
||||
|
||||
|
||||
--
|
||||
-- Name: schema_migration_version_idx; Type: INDEX; Schema: public; Owner: newsbox
|
||||
-- Name: schema_migration_version_idx; Type: INDEX; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
CREATE UNIQUE INDEX schema_migration_version_idx ON public.schema_migration USING btree (version);
|
||||
|
||||
|
||||
--
|
||||
-- Name: users_email_idx; Type: INDEX; Schema: public; Owner: newsbox
|
||||
-- Name: users_email_idx; Type: INDEX; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
CREATE UNIQUE INDEX users_email_idx ON public.users USING btree (email);
|
||||
|
||||
|
||||
--
|
||||
-- Name: users_username_idx; Type: INDEX; Schema: public; Owner: newsbox
|
||||
-- Name: users_username_idx; Type: INDEX; Schema: public; Owner: postgres
|
||||
--
|
||||
|
||||
CREATE UNIQUE INDEX users_username_idx ON public.users USING btree (username);
|
||||
|
|
|
|||
131
mill/mill.go
131
mill/mill.go
|
|
@ -1,131 +0,0 @@
|
|||
package mill
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"log"
|
||||
"newsbox/internal"
|
||||
"newsbox/mailers"
|
||||
"newsbox/models"
|
||||
"time"
|
||||
|
||||
"github.com/btubbs/pgq"
|
||||
)
|
||||
|
||||
type Handler interface {
|
||||
Perform(payload []byte) (err error)
|
||||
}
|
||||
|
||||
type IncomingEmailHandler struct {
|
||||
Handler
|
||||
Worker *pgq.Worker
|
||||
Db *sql.DB
|
||||
}
|
||||
|
||||
type OutgoingEmailHandler struct {
|
||||
Handler
|
||||
Worker *pgq.Worker
|
||||
Db *sql.DB
|
||||
}
|
||||
|
||||
type WelcomeEmailHandler struct {
|
||||
Handler
|
||||
Worker *pgq.Worker
|
||||
Db *sql.DB
|
||||
}
|
||||
|
||||
type EmailChangedEmailHandler struct {
|
||||
Handler
|
||||
Worker *pgq.Worker
|
||||
Db *sql.DB
|
||||
}
|
||||
|
||||
type WelcomeEmailJob struct {
|
||||
Recipient string
|
||||
VerificationURL string
|
||||
}
|
||||
|
||||
type EmailAddressChangeEmailJob struct {
|
||||
Recipient string
|
||||
VerificationURL string
|
||||
}
|
||||
|
||||
// Process processes incoming models.Message payloads
|
||||
func (h *IncomingEmailHandler) Process(payload []byte) (err error) {
|
||||
var msg models.Message
|
||||
err = json.Unmarshal(payload, &msg)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
var owner models.User
|
||||
err = models.DB.Eager().Find(&owner, msg.OwnerID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Re-address the email to the user's email address
|
||||
msg.To = owner.Email.Interface().(string)
|
||||
|
||||
// Determine time to deliver
|
||||
tzOffset := owner.Preferences.TimeZoneUtcOffset
|
||||
|
||||
t := internal.TimeInZone(tzOffset)
|
||||
et := time.Date(t.Year(), t.Month(), t.Day(), owner.Preferences.EmailHourOfDay, 0, 0, 0, t.Location())
|
||||
diff := et.Sub(t)
|
||||
|
||||
// The message should be delivered later today
|
||||
if diff > 0 {
|
||||
et = t.Add(diff)
|
||||
} else { // The message should be delivered tomorrow
|
||||
et = time.Date(t.Year(), t.Month(), t.Day()+1, owner.Preferences.EmailHourOfDay, 0, 0, 0, t.Location())
|
||||
}
|
||||
|
||||
log.Println("Queueing for:", et)
|
||||
log.Println("Assumed location:", t.Location())
|
||||
job, _ := json.MarshalIndent(&msg, "", "\t")
|
||||
_, err = h.Worker.EnqueueJob("outgoing_email", job, pgq.After(et))
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// Process processes outgoing models.Message payloads
|
||||
func (w *OutgoingEmailHandler) Process(payload []byte) (err error) {
|
||||
var msg models.Message
|
||||
err = json.Unmarshal(payload, &msg)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
var owner models.User
|
||||
err = models.DB.Find(&owner, msg.OwnerID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = mailers.ForwardMessage(owner.Email.Interface().(string), &msg)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// Process processes new welcome email jobs
|
||||
func (h *WelcomeEmailHandler) Process(payload []byte) (err error) {
|
||||
var job WelcomeEmailJob
|
||||
err = json.Unmarshal(payload, &job)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = mailers.SendWelcomeEmail(job.Recipient, job.VerificationURL)
|
||||
return
|
||||
}
|
||||
|
||||
// Process processes changes to user account email addresses
|
||||
func (h *EmailChangedEmailHandler) Process(payload []byte) (err error) {
|
||||
var job EmailAddressChangeEmailJob
|
||||
err = json.Unmarshal(payload, &job)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = mailers.SendEmailAddressChangedEmail(job.Recipient, job.VerificationURL)
|
||||
return
|
||||
}
|
||||
|
|
@ -1,62 +1,127 @@
|
|||
package workers
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"newsbox/mill"
|
||||
"os"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log"
|
||||
"newsbox/internal"
|
||||
"newsbox/mailers"
|
||||
"newsbox/models"
|
||||
"time"
|
||||
|
||||
"github.com/btubbs/pgq"
|
||||
"github.com/gobuffalo/envy"
|
||||
"github.com/acaloiaro/neoq"
|
||||
)
|
||||
|
||||
var W *pgq.Worker
|
||||
var nq neoq.Neoq
|
||||
|
||||
func init() {
|
||||
dbUrl, err := envy.MustGet("DATABASE_URL")
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to get DATABASE_URL")
|
||||
}
|
||||
db, err := sql.Open("postgres", dbUrl)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
func Start() {
|
||||
log.Println("Firing up workers")
|
||||
dbURL := internal.DatabaseUrl()
|
||||
|
||||
W = pgq.NewWorker(db)
|
||||
incomingHandler := mill.IncomingEmailHandler{Db: db, Worker: W}
|
||||
err = W.RegisterQueue("incoming_email", incomingHandler.Process)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to register 'incoming_email' queue handler")
|
||||
panic(err)
|
||||
}
|
||||
|
||||
outgoingHandler := mill.OutgoingEmailHandler{Db: db, Worker: W}
|
||||
err = W.RegisterQueue("outgoing_email", outgoingHandler.Process)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to register 'incoming_email' queue handler")
|
||||
panic(err)
|
||||
}
|
||||
|
||||
welcomeHandler := mill.WelcomeEmailHandler{Db: db, Worker: W}
|
||||
err = W.RegisterQueue("welcome_emails", welcomeHandler.Process)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to register 'welcome_emails' queue handler")
|
||||
panic(err)
|
||||
}
|
||||
|
||||
passwordChangeHandler := mill.EmailChangedEmailHandler{Db: db, Worker: W}
|
||||
err = W.RegisterQueue("email_address_changes", passwordChangeHandler.Process)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to register 'password_changes' queue handler")
|
||||
panic(err)
|
||||
}
|
||||
|
||||
// Start up the worker
|
||||
go func() {
|
||||
err = W.Run()
|
||||
nq, _ = neoq.New(dbURL)
|
||||
nq.Listen("incoming_email", neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
j, err := neoq.JobFromContext(ctx)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
return
|
||||
}
|
||||
}()
|
||||
|
||||
var msg models.Message
|
||||
err = models.DB.Eager().Find(&msg, j.Payload["message_id"])
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
var owner models.User
|
||||
err = models.DB.Eager().Find(&owner, msg.OwnerID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Re-address the email to the user's email address
|
||||
msg.To = owner.Email.Interface().(string)
|
||||
|
||||
// Determine time to deliver
|
||||
tzOffset := owner.Preferences.TimeZoneUtcOffset
|
||||
|
||||
t := internal.TimeInZone(tzOffset)
|
||||
et := time.Date(t.Year(), t.Month(), t.Day(), owner.Preferences.EmailHourOfDay, 0, 0, 0, t.Location())
|
||||
diff := et.Sub(t)
|
||||
|
||||
// The message should be delivered later today
|
||||
if diff > 0 {
|
||||
et = t.Add(diff)
|
||||
} else { // The message should be delivered tomorrow
|
||||
et = time.Date(t.Year(), t.Month(), t.Day()+1, owner.Preferences.EmailHourOfDay, 0, 0, 0, t.Location())
|
||||
}
|
||||
|
||||
msgJSON, _ := json.Marshal(msg)
|
||||
|
||||
log.Println("Queueing for:", et)
|
||||
log.Println("Assumed location:", t.Location())
|
||||
nq, _ := neoq.New(dbURL)
|
||||
_, err = nq.Enqueue(neoq.Job{
|
||||
Queue: "outgoing_email",
|
||||
Payload: map[string]any{
|
||||
"message": string(msgJSON),
|
||||
},
|
||||
})
|
||||
|
||||
return
|
||||
}))
|
||||
|
||||
nq.Listen("welcome_emails", neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
j, err := neoq.JobFromContext(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
recipient := j.Payload["recipient"].(string)
|
||||
verificationURL := j.Payload["verification_url"].(string)
|
||||
|
||||
log.Println("sending welcome email to:", recipient)
|
||||
err = mailers.SendWelcomeEmail(recipient, verificationURL)
|
||||
log.Println("What's my error here?", err)
|
||||
|
||||
return
|
||||
|
||||
}))
|
||||
|
||||
nq.Listen("outgoing_email", neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
var msg models.Message
|
||||
j, err := neoq.JobFromContext(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = json.Unmarshal([]byte(j.Payload["message"].(string)), &msg)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
var owner models.User
|
||||
err = models.DB.Find(&owner, msg.OwnerID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = mailers.ForwardMessage(owner.Email.Interface().(string), &msg)
|
||||
|
||||
return
|
||||
|
||||
}))
|
||||
|
||||
nq.Listen("email_verifications", neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
j, err := neoq.JobFromContext(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
recipient := j.Payload["recipient"].(string)
|
||||
verificationURL := j.Payload["verification_url"].(string)
|
||||
|
||||
err = mailers.SendEmailAddressChangedEmail(recipient, verificationURL)
|
||||
return
|
||||
|
||||
}))
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue