134 lines
6.0 KiB
Go
134 lines
6.0 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"git.sendico.io/tech/sxchnge/notifications/model"
|
|
"git.sendico.io/tech/sxchnge/shared/database/repository"
|
|
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
|
|
ri "git.sendico.io/tech/sxchnge/shared/database/repository/index"
|
|
"git.sendico.io/tech/sxchnge/shared/database/template"
|
|
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
|
mlogger "git.sendico.io/tech/sxchnge/shared/logging"
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|
"go.mongodb.org/mongo-driver/v2/mongo"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
type Store struct {
|
|
template.DBImp[*model.Notification]
|
|
}
|
|
|
|
func New(logger mlogger.Logger, db *mongo.Database) (*Store, error) {
|
|
s := &Store{*template.Create[*model.Notification](mlogger.Named(logger, "store"), "notifications", db)}
|
|
ttl := int32(0)
|
|
for _, index := range []*ri.Definition{
|
|
{Name: "unique_caller_key", Keys: []ri.Key{{Field: "caller", Sort: ri.Asc}, {Field: "key", Sort: ri.Asc}}, Unique: true},
|
|
{Name: "pending_dispatch", Keys: []ri.Key{{Field: "state", Sort: ri.Asc}, {Field: "next_attempt_at", Sort: ri.Asc}}},
|
|
{Name: "pending_expiry", Keys: []ri.Key{{Field: "state", Sort: ri.Asc}, {Field: "expires_at", Sort: ri.Asc}}},
|
|
{Name: "expired_lease", Keys: []ri.Key{{Field: "state", Sort: ri.Asc}, {Field: "lease_until", Sort: ri.Asc}}},
|
|
{Name: "purge_notifications", Keys: []ri.Key{{Field: "purge_at", Sort: ri.Asc}}, TTL: &ttl},
|
|
} {
|
|
if err := s.Repository.CreateIndex(index); err != nil {
|
|
return nil, wrap(err)
|
|
}
|
|
}
|
|
return s, nil
|
|
}
|
|
func wrap(err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
if errors.Is(err, merrors.ErrNoData) || errors.Is(err, merrors.ErrDataConflict) {
|
|
return err
|
|
}
|
|
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "notification storage unavailable")
|
|
}
|
|
func (s *Store) Find(ctx context.Context, caller, key string) (*model.Notification, error) {
|
|
var n model.Notification
|
|
err := s.FindOne(ctx, repository.Query().And(repository.Filter("caller", caller), repository.Filter("key", key)), &n)
|
|
return &n, wrap(err)
|
|
}
|
|
func (s *Store) Get(ctx context.Context, caller, id string) (*model.Notification, error) {
|
|
oid, err := bson.ObjectIDFromHex(id)
|
|
if err != nil {
|
|
return nil, merrors.InvalidArgument("invalid notification ID")
|
|
}
|
|
var n model.Notification
|
|
err = s.FindOne(ctx, repository.Query().And(repository.Filter("caller", caller), repository.IDFilter(oid)), &n)
|
|
return &n, wrap(err)
|
|
}
|
|
func (s *Store) Create(ctx context.Context, n *model.Notification) (*model.Notification, error) {
|
|
err := s.DBImp.Create(ctx, n)
|
|
if errors.Is(err, merrors.ErrDataConflict) {
|
|
return s.Find(ctx, n.Caller, n.Key)
|
|
}
|
|
return n, wrap(err)
|
|
}
|
|
func pending(now time.Time) builder.Query {
|
|
return repository.Query().And(repository.Query().In(repository.Field("state"), model.Queued, model.RetryWait), repository.Query().Comparison(repository.Field("next_attempt_at"), builder.Lte, now))
|
|
}
|
|
|
|
// Expire pending messages independently of bus availability, then fence lost workers.
|
|
// A lost worker may already have sent its message. Do not automatically replay it.
|
|
func (s *Store) Recover(ctx context.Context, now time.Time) error {
|
|
expired := repository.Query().And(repository.Query().In(repository.Field("state"), model.Queued, model.RetryWait), repository.Query().Comparison(repository.Field("expires_at"), builder.Lte, now))
|
|
expirePatch := repository.Patch().Set(repository.Field("state"), model.Expired).Set(repository.Field("failure_code"), "expired").Set(repository.Field("updatedAt"), now).Set(repository.Field("email"), model.Email{})
|
|
count, err := s.PatchMany(ctx, expired, expirePatch)
|
|
if err != nil {
|
|
return wrap(err)
|
|
}
|
|
if count > 0 {
|
|
s.Logger.Info("Pending notifications expired", zap.Int("count", count))
|
|
}
|
|
query := repository.Query().And(repository.Filter("state", model.Sending), repository.Query().Comparison(repository.Field("lease_until"), builder.Lte, now))
|
|
patch := repository.Patch().Set(repository.Field("state"), model.Unknown).Set(repository.Field("failure_code"), "worker_outcome_unknown").Set(repository.Field("updatedAt"), now).Set(repository.Field("email"), model.Email{})
|
|
count, err = s.PatchMany(ctx, query, patch)
|
|
if err == nil && count > 0 {
|
|
s.Logger.Warn("Notification workers lost their leases", zap.Int("count", count))
|
|
}
|
|
return wrap(err)
|
|
}
|
|
func (s *Store) Claim(ctx context.Context, id string, now time.Time) (*model.Notification, error) {
|
|
oid, err := bson.ObjectIDFromHex(id)
|
|
if err != nil {
|
|
return nil, merrors.NoData("invalid job ID")
|
|
}
|
|
var n model.Notification
|
|
query := repository.Query().And(repository.IDFilter(oid), pending(now))
|
|
patch := repository.Patch().Set(repository.Field("state"), model.Sending).Set(repository.Field("updatedAt"), now).Set(repository.Field("lease_until"), now.Add(time.Minute)).Inc(repository.Field("attempts"), 1)
|
|
matched, err := s.Repository.PatchOneByFilter(ctx, query, patch, &n)
|
|
if err != nil {
|
|
return nil, wrap(err)
|
|
}
|
|
if !matched {
|
|
return nil, merrors.NoData("job is not pending")
|
|
}
|
|
return &n, nil
|
|
}
|
|
func (s *Store) Finish(ctx context.Context, n *model.Notification) error {
|
|
query := repository.Query().And(repository.IDFilter(n.ID), repository.Filter("state", model.Sending), repository.Filter("attempts", n.Attempts), repository.Filter("lease_until", n.LeaseUntil))
|
|
patch := repository.Patch().Set(repository.Field("state"), n.State).Set(repository.Field("failure_code"), n.FailureCode).Set(repository.Field("updatedAt"), n.UpdatedAt).Set(repository.Field("next_attempt_at"), n.NextAttemptAt)
|
|
if n.State != model.RetryWait {
|
|
patch.Set(repository.Field("email"), model.Email{})
|
|
}
|
|
count, err := s.PatchMany(ctx, query, patch)
|
|
if err != nil {
|
|
return wrap(err)
|
|
}
|
|
if count != 1 {
|
|
return merrors.DataConflict("delivery lease lost")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Acceptance and outbox work are one document, so no dual-write transaction is needed.
|
|
func (s *Store) Due(ctx context.Context, now time.Time) ([]model.Notification, error) {
|
|
limit := int64(100)
|
|
var jobs []model.Notification
|
|
err := s.Repository.FindAllByFilter(ctx, pending(now).Sort(repository.Field("next_attempt_at"), true).Limit(&limit), &jobs)
|
|
return jobs, wrap(err)
|
|
}
|