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")) duplicate := &Binding{Namespace: b.Namespace} duplicate.SetID(bson.NewObjectID()) require.NoError(t, store.Create(ctx, duplicate)) claimedDuplicate, err := store.Claim(ctx, duplicate.ID, duplicate.Namespace) require.NoError(t, err) require.True(t, claimedDuplicate) require.ErrorIs(t, store.Bind(ctx, duplicate.ID, duplicate.Namespace, "subject"), merrors.ErrDataConflict) 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) }