Connect sign-up to Sendico Facts intake (#46)
This commit is contained in:
Stephan D committed
1 parent
30f0455122
commit
a584c8189a
40 files changed
+1616
-35
No files matched your search
@@ -0,0 +1,9 @@
|
||||
package identity
|
||||
|
||||
import (
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promauto"
|
||||
)
|
||||
|
||||
var requests = promauto.NewCounterVec(prometheus.CounterOpts{Name: "sxchnge_signup_facts_total", Help: "Sign-up facts handoff attempts; acceptance is not verification."}, []string{"outcome"})
|
||||
var duration = promauto.NewHistogram(prometheus.HistogramOpts{Name: "sxchnge_signup_facts_duration_seconds", Help: "Sign-up facts handoff latency including retries."})
|
||||
@@ -0,0 +1,39 @@
|
||||
// Package identity owns the durable sign-up facts handoff, not KYC decisions.
|
||||
package identity
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"git.sendico.io/tech/sxchnge/shared/database/record"
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
)
|
||||
|
||||
// Facts is a snapshot of local sign-up assertions. It never contains credentials.
|
||||
type Facts struct {
|
||||
FullName string `bson:"fullName"`
|
||||
Email string `bson:"email"`
|
||||
CustomerRef string `bson:"customerRef"`
|
||||
}
|
||||
|
||||
type Receipt struct {
|
||||
SubmissionRef string `bson:"submissionRef"`
|
||||
SubjectRef string `bson:"subjectRef"`
|
||||
AcceptedAt time.Time `bson:"acceptedAt"`
|
||||
}
|
||||
|
||||
// Binding uses the local customer ID as its primary key. Namespace is an opaque
|
||||
// adapter configuration fingerprint; no platform tenancy enters the customer model.
|
||||
type Binding struct {
|
||||
storable.Base `bson:",inline"`
|
||||
Namespace string `bson:"namespace"`
|
||||
RegistrationRequest string `bson:"registrationRequest"`
|
||||
Facts Facts `bson:"facts"`
|
||||
Registering bool `bson:"registering"`
|
||||
RegisteringAt time.Time `bson:"registeringAt"`
|
||||
SubjectRef string `bson:"subjectRef"`
|
||||
Receipt *Receipt `bson:"receipt,omitempty"`
|
||||
}
|
||||
|
||||
func (*Binding) Collection() string { return "signup_facts" }
|
||||
|
||||
func sourceRef(id bson.ObjectID) string { return "sxchnge:signup:" + id.Hex() }
|
||||
@@ -0,0 +1,132 @@
|
||||
package identity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
customer "git.sendico.io/tech/sxchnge/product/customers/model"
|
||||
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
||||
mlogger "git.sendico.io/tech/sxchnge/shared/logging"
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type Provider interface {
|
||||
Namespace() string
|
||||
Register(context.Context, string) (string, error)
|
||||
Submit(context.Context, string, string, Facts) (Receipt, error)
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
Store Store
|
||||
Provider Provider
|
||||
Logger mlogger.Logger
|
||||
}
|
||||
|
||||
func unavailable() error { return merrors.ServiceUnavailable("sign-up facts unavailable") }
|
||||
|
||||
// Ensure completes only after a provider receipt has been persisted. Registration
|
||||
// has no idempotency contract, so its durable claim is never automatically expired.
|
||||
func (s *Service) Ensure(ctx context.Context, a *customer.Account) (result error) {
|
||||
if a == nil || a.ID.IsZero() {
|
||||
return merrors.InvalidArgument("customer required")
|
||||
}
|
||||
started := time.Now()
|
||||
requestRef := ""
|
||||
outcome := "unavailable"
|
||||
defer func() {
|
||||
requests.WithLabelValues(outcome).Inc()
|
||||
duration.Observe(time.Since(started).Seconds())
|
||||
fields := []zap.Field{zap.String("account_ref", a.ID.Hex()), zap.String("request_ref", requestRef), zap.String("outcome", outcome)}
|
||||
if result != nil {
|
||||
s.Logger.Warn("Sign-up facts handoff incomplete", fields...)
|
||||
} else {
|
||||
s.Logger.Info("Sign-up facts handoff completed", fields...)
|
||||
}
|
||||
}()
|
||||
namespace := s.Provider.Namespace()
|
||||
b, err := s.Store.Get(ctx, a.ID)
|
||||
if errors.Is(err, merrors.ErrNoData) {
|
||||
b = &Binding{Namespace: namespace, RegistrationRequest: bson.NewObjectID().Hex(), Facts: Facts{FullName: a.Name, Email: a.Login, CustomerRef: a.ID.Hex()}}
|
||||
b.SetID(a.ID)
|
||||
if err = s.Store.Create(ctx, b); err != nil {
|
||||
if !errors.Is(err, merrors.ErrDataConflict) {
|
||||
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "persist sign-up facts")
|
||||
}
|
||||
b, err = s.Store.Get(ctx, a.ID)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "load sign-up facts")
|
||||
}
|
||||
requestRef = b.RegistrationRequest
|
||||
if b.Namespace != namespace {
|
||||
outcome = "attention_required"
|
||||
return merrors.FailedPrecondition("sign-up provider binding changed")
|
||||
}
|
||||
if b.Receipt != nil {
|
||||
outcome = "replayed"
|
||||
return nil
|
||||
}
|
||||
if b.SubjectRef == "" {
|
||||
claimed, err := s.Store.Claim(ctx, a.ID, namespace)
|
||||
if err != nil {
|
||||
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "claim individual registration")
|
||||
}
|
||||
if !claimed {
|
||||
// Another request may have completed while this request was claiming.
|
||||
b, err = s.Store.Get(ctx, a.ID)
|
||||
if err != nil {
|
||||
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "reload sign-up facts")
|
||||
}
|
||||
if b.SubjectRef == "" {
|
||||
if b.RegisteringAt.Add(30 * time.Second).After(time.Now()) {
|
||||
return unavailable()
|
||||
}
|
||||
outcome = "attention_required"
|
||||
return merrors.FailedPrecondition("individual registration requires reconciliation")
|
||||
}
|
||||
} else {
|
||||
subject, err := s.Provider.Register(ctx, b.RegistrationRequest)
|
||||
if err != nil {
|
||||
outcome = "attention_required"
|
||||
return merrors.OpaqueWrap(merrors.ErrFailedPrecondition, err, "individual registration requires reconciliation")
|
||||
}
|
||||
// Persist even if the browser left while Sendico accepted registration.
|
||||
saveCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
||||
err = s.Store.Bind(saveCtx, a.ID, namespace, subject)
|
||||
cancel()
|
||||
if err != nil {
|
||||
outcome = "attention_required"
|
||||
s.Logger.Warn("Registered individual binding needs reconciliation", zap.String("account_ref", a.ID.Hex()), zap.String("subject_ref", subject))
|
||||
return merrors.OpaqueWrap(merrors.ErrFailedPrecondition, err, "persist individual binding")
|
||||
}
|
||||
b.SubjectRef = subject
|
||||
}
|
||||
}
|
||||
receipt, err := s.Provider.Submit(ctx, b.SubjectRef, sourceRef(a.ID), b.Facts)
|
||||
if err != nil {
|
||||
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "submit sign-up facts")
|
||||
}
|
||||
if receipt.SubjectRef != b.SubjectRef || receipt.SubmissionRef == "" || receipt.AcceptedAt.IsZero() {
|
||||
return unavailable()
|
||||
}
|
||||
if err = s.Store.Accept(ctx, a.ID, namespace, receipt); err != nil {
|
||||
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "persist facts receipt")
|
||||
}
|
||||
outcome = "accepted"
|
||||
return nil
|
||||
}
|
||||
|
||||
// Ready checks confirmed intake without external I/O inside activation transactions.
|
||||
func (s *Service) Ready(ctx context.Context, id bson.ObjectID) error {
|
||||
b, err := s.Store.Get(ctx, id)
|
||||
if err != nil {
|
||||
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "load facts receipt")
|
||||
}
|
||||
if b.Namespace != s.Provider.Namespace() || b.Receipt == nil {
|
||||
return unavailable()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
package identity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
customer "git.sendico.io/tech/sxchnge/product/customers/model"
|
||||
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
)
|
||||
|
||||
type memoryStore struct {
|
||||
mu sync.Mutex
|
||||
rows map[bson.ObjectID]Binding
|
||||
bindErr, acceptErr error
|
||||
}
|
||||
|
||||
func (s *memoryStore) Get(_ context.Context, id bson.ObjectID) (*Binding, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
b, ok := s.rows[id]
|
||||
if !ok {
|
||||
return nil, merrors.NoData("missing")
|
||||
}
|
||||
return &b, nil
|
||||
}
|
||||
func (s *memoryStore) Create(_ context.Context, b *Binding) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if _, ok := s.rows[b.ID]; ok {
|
||||
return merrors.DataConflict("duplicate")
|
||||
}
|
||||
s.rows[b.ID] = *b
|
||||
return nil
|
||||
}
|
||||
func (s *memoryStore) Claim(_ context.Context, id bson.ObjectID, namespace string) (bool, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
b := s.rows[id]
|
||||
if b.Namespace != namespace || b.Registering || b.SubjectRef != "" {
|
||||
return false, nil
|
||||
}
|
||||
b.Registering = true
|
||||
b.RegisteringAt = time.Now()
|
||||
s.rows[id] = b
|
||||
return true, nil
|
||||
}
|
||||
func (s *memoryStore) Bind(_ context.Context, id bson.ObjectID, namespace, subject string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.bindErr != nil {
|
||||
return s.bindErr
|
||||
}
|
||||
b := s.rows[id]
|
||||
b.SubjectRef = subject
|
||||
b.Registering = false
|
||||
s.rows[id] = b
|
||||
return nil
|
||||
}
|
||||
func (s *memoryStore) Accept(_ context.Context, id bson.ObjectID, namespace string, r Receipt) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.acceptErr != nil {
|
||||
return s.acceptErr
|
||||
}
|
||||
b := s.rows[id]
|
||||
b.Receipt = &r
|
||||
s.rows[id] = b
|
||||
return nil
|
||||
}
|
||||
|
||||
type providerStub struct {
|
||||
registrations, submissions int
|
||||
registerErr, submitErr error
|
||||
snapshots []Facts
|
||||
sources []string
|
||||
entered, release chan struct{}
|
||||
}
|
||||
|
||||
func (*providerStub) Namespace() string { return "provider-1" }
|
||||
func (p *providerStub) Register(context.Context, string) (string, error) {
|
||||
p.registrations++
|
||||
if p.entered != nil {
|
||||
close(p.entered)
|
||||
<-p.release
|
||||
}
|
||||
return "subject-1", p.registerErr
|
||||
}
|
||||
func (p *providerStub) Submit(_ context.Context, subject, source string, f Facts) (Receipt, error) {
|
||||
p.submissions++
|
||||
p.snapshots = append(p.snapshots, f)
|
||||
p.sources = append(p.sources, source)
|
||||
return Receipt{SubjectRef: subject, SubmissionRef: "submission-1", AcceptedAt: time.Now()}, p.submitErr
|
||||
}
|
||||
func fixture() (*Service, *memoryStore, *providerStub, *customer.Account) {
|
||||
store := &memoryStore{rows: map[bson.ObjectID]Binding{}}
|
||||
provider := &providerStub{}
|
||||
a := &customer.Account{}
|
||||
a.ID = bson.NewObjectID()
|
||||
a.Name = "Ada Lovelace"
|
||||
a.Login = "ada@example.com"
|
||||
a.Password = "never transmit"
|
||||
return &Service{Store: store, Provider: provider, Logger: zap.NewNop()}, store, provider, a
|
||||
}
|
||||
|
||||
func TestIntakeRetryUsesPersistedSubjectSnapshotAndStableSource(t *testing.T) {
|
||||
s, store, p, a := fixture()
|
||||
p.submitErr = merrors.ServiceUnavailable("upstream")
|
||||
require.Error(t, s.Ensure(context.Background(), a))
|
||||
require.Error(t, s.Ready(context.Background(), a.ID))
|
||||
require.Equal(t, 1, p.registrations)
|
||||
b, err := store.Get(context.Background(), a.ID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "subject-1", b.SubjectRef)
|
||||
require.Nil(t, b.Receipt)
|
||||
a.Name = "changed retry"
|
||||
p.submitErr = nil
|
||||
require.NoError(t, s.Ensure(context.Background(), a))
|
||||
require.NoError(t, s.Ready(context.Background(), a.ID))
|
||||
require.Equal(t, 1, p.registrations)
|
||||
require.Equal(t, 2, p.submissions)
|
||||
require.Equal(t, p.snapshots[0], p.snapshots[1])
|
||||
require.Equal(t, "Ada Lovelace", p.snapshots[1].FullName)
|
||||
require.Equal(t, p.sources[0], p.sources[1])
|
||||
require.NoError(t, s.Ensure(context.Background(), a))
|
||||
require.Equal(t, 2, p.submissions)
|
||||
other := a.Copy()
|
||||
other.ID = bson.NewObjectID()
|
||||
require.Error(t, s.Ready(context.Background(), other.ID))
|
||||
}
|
||||
func TestLostAcceptanceReceiptIsSafelyReplayed(t *testing.T) {
|
||||
s, store, p, a := fixture()
|
||||
store.acceptErr = merrors.ServiceUnavailable("storage")
|
||||
require.Error(t, s.Ensure(context.Background(), a))
|
||||
store.acceptErr = nil
|
||||
require.NoError(t, s.Ensure(context.Background(), a))
|
||||
require.Equal(t, 1, p.registrations)
|
||||
require.Equal(t, 2, p.submissions)
|
||||
require.Equal(t, p.sources[0], p.sources[1])
|
||||
}
|
||||
func TestAmbiguousRegistrationNeverAutomaticallyRegistersAgain(t *testing.T) {
|
||||
for _, bindFailure := range []bool{false, true} {
|
||||
t.Run(map[bool]string{false: "request", true: "binding"}[bindFailure], func(t *testing.T) {
|
||||
s, store, p, a := fixture()
|
||||
if bindFailure {
|
||||
store.bindErr = merrors.ServiceUnavailable("storage")
|
||||
} else {
|
||||
p.registerErr = merrors.ServiceUnavailable("lost response")
|
||||
}
|
||||
require.ErrorIs(t, s.Ensure(context.Background(), a), merrors.ErrFailedPrecondition)
|
||||
b := store.rows[a.ID]
|
||||
b.RegisteringAt = time.Now().Add(-time.Minute)
|
||||
store.rows[a.ID] = b
|
||||
store.bindErr = nil
|
||||
p.registerErr = nil
|
||||
require.ErrorIs(t, s.Ensure(context.Background(), a), merrors.ErrFailedPrecondition)
|
||||
require.Equal(t, 1, p.registrations)
|
||||
require.Zero(t, p.submissions)
|
||||
})
|
||||
}
|
||||
}
|
||||
func TestConcurrentRequestCannotRegisterSameCustomerTwice(t *testing.T) {
|
||||
s, _, p, a := fixture()
|
||||
p.entered = make(chan struct{})
|
||||
p.release = make(chan struct{})
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- s.Ensure(context.Background(), a) }()
|
||||
<-p.entered
|
||||
require.ErrorIs(t, s.Ensure(context.Background(), a), merrors.ErrServiceUnavailable)
|
||||
close(p.release)
|
||||
require.NoError(t, <-done)
|
||||
require.Equal(t, 1, p.registrations)
|
||||
require.NoError(t, s.Ensure(context.Background(), a))
|
||||
require.Equal(t, 1, p.submissions)
|
||||
}
|
||||
func TestNamespaceChangeFailsClosed(t *testing.T) {
|
||||
s, store, p, a := fixture()
|
||||
require.NoError(t, s.Ensure(context.Background(), a))
|
||||
b := store.rows[a.ID]
|
||||
b.Namespace = "other-account-or-organization"
|
||||
store.rows[a.ID] = b
|
||||
require.ErrorIs(t, s.Ensure(context.Background(), a), merrors.ErrFailedPrecondition)
|
||||
require.Error(t, s.Ready(context.Background(), a.ID))
|
||||
require.Equal(t, 1, p.submissions)
|
||||
}
|
||||
|
||||
func TestHandoffLogsExcludeAssertionsAndCredentials(t *testing.T) {
|
||||
s, _, p, a := fixture()
|
||||
core, logs := observer.New(zap.InfoLevel)
|
||||
s.Logger = zap.New(core)
|
||||
p.submitErr = merrors.ServiceUnavailable(a.Login + " " + a.Password)
|
||||
require.Error(t, s.Ensure(context.Background(), a))
|
||||
p.submitErr = nil
|
||||
require.NoError(t, s.Ensure(context.Background(), a))
|
||||
require.Len(t, logs.All(), 2)
|
||||
for _, entry := range logs.All() {
|
||||
require.NotContains(t, entry.Message, a.Login)
|
||||
require.NotContains(t, entry.Message, a.Name)
|
||||
fields := entry.ContextMap()
|
||||
require.Equal(t, a.ID.Hex(), fields["account_ref"])
|
||||
require.NotEmpty(t, fields["request_ref"])
|
||||
for _, value := range fields {
|
||||
require.NotEqual(t, a.Login, value)
|
||||
require.NotEqual(t, a.Name, value)
|
||||
require.NotEqual(t, a.Password, value)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package identity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
db "git.sendico.io/tech/sxchnge/shared/database"
|
||||
deps "git.sendico.io/tech/sxchnge/shared/database/dependencies"
|
||||
"git.sendico.io/tech/sxchnge/shared/database/repository"
|
||||
"git.sendico.io/tech/sxchnge/shared/database/source"
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
)
|
||||
|
||||
type Store interface {
|
||||
Get(context.Context, bson.ObjectID) (*Binding, error)
|
||||
Create(context.Context, *Binding) error
|
||||
Claim(context.Context, bson.ObjectID, string) (bool, error)
|
||||
Bind(context.Context, bson.ObjectID, string, string) error
|
||||
Accept(context.Context, bson.ObjectID, string, Receipt) error
|
||||
}
|
||||
|
||||
type mongoStore struct{ records repository.Repository }
|
||||
|
||||
func OpenStore(factory db.Factory) (Store, error) {
|
||||
d, err := deps.ResolveDependencies(factory)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &mongoStore{records: repository.CreateRepository(source.Majority(d.Database), "signup_facts")}, nil
|
||||
}
|
||||
|
||||
func (s *mongoStore) Get(ctx context.Context, id bson.ObjectID) (*Binding, error) {
|
||||
b := &Binding{}
|
||||
err := s.records.Get(ctx, id, b)
|
||||
return b, err
|
||||
}
|
||||
func (s *mongoStore) Create(ctx context.Context, b *Binding) error {
|
||||
return s.records.Insert(ctx, b, nil)
|
||||
}
|
||||
func scoped(id bson.ObjectID, namespace string) repository.FilterQuery {
|
||||
return repository.IDFilter(id).Filter(repository.Field("namespace"), namespace)
|
||||
}
|
||||
func (s *mongoStore) Claim(ctx context.Context, id bson.ObjectID, namespace string) (bool, error) {
|
||||
query := scoped(id, namespace).Filter(repository.Field("registering"), false).Filter(repository.Field("subjectRef"), "")
|
||||
return s.records.PatchOneByFilter(ctx, query, repository.Patch().Set(repository.Field("registering"), true).Set(repository.Field("registeringAt"), time.Now().UTC()), &Binding{})
|
||||
}
|
||||
func (s *mongoStore) Bind(ctx context.Context, id bson.ObjectID, namespace, subject string) error {
|
||||
query := scoped(id, namespace).Filter(repository.Field("registering"), true).Filter(repository.Field("subjectRef"), "")
|
||||
matched, err := s.records.PatchOneByFilter(ctx, query, repository.Patch().Set(repository.Field("subjectRef"), subject).Set(repository.Field("registering"), false), &Binding{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !matched {
|
||||
return unavailable()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (s *mongoStore) Accept(ctx context.Context, id bson.ObjectID, namespace string, receipt Receipt) error {
|
||||
query := scoped(id, namespace).Filter(repository.Field("subjectRef"), receipt.SubjectRef)
|
||||
matched, err := s.records.PatchOneByFilter(ctx, query, repository.Patch().Set(repository.Field("receipt"), receipt), &Binding{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !matched {
|
||||
return unavailable()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package identity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/url"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
db "git.sendico.io/tech/sxchnge/shared/database"
|
||||
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
||||
sharedmodel "git.sendico.io/tech/sxchnge/shared/model"
|
||||
"github.com/stretchr/testify/require"
|
||||
"github.com/testcontainers/testcontainers-go/modules/mongodb"
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
func TestDurableBindingClaimsAndIsolation(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
||||
defer cancel()
|
||||
container, err := mongodb.Run(ctx, "mongo:latest", mongodb.WithReplicaSet("rs"))
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, container.Terminate(context.Background())) })
|
||||
uri, err := container.ConnectionString(ctx)
|
||||
require.NoError(t, err)
|
||||
u, err := url.Parse(uri)
|
||||
require.NoError(t, err)
|
||||
q := u.Query()
|
||||
q.Set("directConnection", "true")
|
||||
// The handoff store must override unsafe caller defaults before claiming
|
||||
// ownership of a non-idempotent external operation.
|
||||
q.Set("w", "0")
|
||||
q.Set("readPreference", "secondary")
|
||||
u.RawQuery = q.Encode()
|
||||
factory, err := db.NewConnection(zap.NewNop(), &db.Config{Driver: db.Mongo, Settings: sharedmodel.SettingsT{"uri": u.String(), "database": "facts"}})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(factory.CloseConnection)
|
||||
store, err := OpenStore(factory)
|
||||
require.NoError(t, err)
|
||||
b := &Binding{Namespace: "configured-source", Facts: Facts{FullName: "Ada", Email: "ada@example.com"}}
|
||||
b.SetID(bson.NewObjectID())
|
||||
require.NoError(t, store.Create(ctx, b))
|
||||
require.ErrorIs(t, store.Create(ctx, b), merrors.ErrDataConflict)
|
||||
var wg sync.WaitGroup
|
||||
var wins atomic.Int32
|
||||
for range 12 {
|
||||
wg.Go(func() {
|
||||
claimed, err := store.Claim(ctx, b.ID, b.Namespace)
|
||||
require.NoError(t, err)
|
||||
if claimed {
|
||||
wins.Add(1)
|
||||
}
|
||||
})
|
||||
}
|
||||
wg.Wait()
|
||||
require.EqualValues(t, 1, wins.Load())
|
||||
stored, err := store.Get(ctx, b.ID)
|
||||
require.NoError(t, err)
|
||||
require.True(t, stored.Registering)
|
||||
require.False(t, stored.RegisteringAt.IsZero())
|
||||
require.Error(t, store.Bind(ctx, b.ID, "other-source", "subject"))
|
||||
require.NoError(t, store.Bind(ctx, b.ID, b.Namespace, "subject"))
|
||||
require.Error(t, store.Bind(ctx, b.ID, b.Namespace, "different-subject"))
|
||||
receipt := Receipt{SubjectRef: "other-subject", SubmissionRef: "submission", AcceptedAt: time.Now().UTC().Truncate(time.Millisecond)}
|
||||
require.Error(t, store.Accept(ctx, b.ID, b.Namespace, receipt))
|
||||
receipt.SubjectRef = "subject"
|
||||
require.NoError(t, store.Accept(ctx, b.ID, b.Namespace, receipt))
|
||||
// A separately opened store sees the durable receipt and cannot reclaim registration.
|
||||
reopened, err := OpenStore(factory)
|
||||
require.NoError(t, err)
|
||||
stored, err = reopened.Get(ctx, b.ID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, receipt, *stored.Receipt)
|
||||
claimed, err := reopened.Claim(ctx, b.ID, b.Namespace)
|
||||
require.NoError(t, err)
|
||||
require.False(t, claimed)
|
||||
_, err = reopened.Get(ctx, bson.NewObjectID())
|
||||
require.ErrorIs(t, err, merrors.ErrNoData)
|
||||
}
|
||||
Reference in new issue
Block a user