sxchnge/backend/edge/shared/storage/organization/provisioning.go

package organization

import (
	"context"
	"errors"
	"time"

	"git.sendico.io/tech/sxchnge/bff/pkg/db/repository"
	"git.sendico.io/tech/sxchnge/bff/pkg/db/repository/builder"
	"git.sendico.io/tech/sxchnge/bff/pkg/db/storable"
	"git.sendico.io/tech/sxchnge/bff/pkg/merrors"
	"git.sendico.io/tech/sxchnge/bff/pkg/model"
	"git.sendico.io/tech/sxchnge/bff/pkg/mutil/mzap"
	"go.mongodb.org/mongo-driver/v2/bson"
	"go.uber.org/zap"
)

// AcquireProvisioningLease claims an organization that is still being
// provisioned. The update is atomic: it only matches a pending organization
// that carries a provisioning record and whose lease has expired or is already
// owned by this attempt, so two concurrent attempts cannot provision the same
// organization. Requiring the pending state here (not only in the caller's
// earlier read) closes the race with a concurrent finalization: an organization
// that just went active stops matching even though its lease fields are gone.
func (db *OrganizationDB) AcquireProvisioningLease(
	ctx context.Context,
	organizationRef bson.ObjectID,
	leaseOwner string,
	now time.Time,
	expiresAt time.Time,
) (bool, error) {
	// A freshly created attempt carries no lease yet, so "missing" counts as free
	// alongside "expired"; otherwise the first attempt would block itself.
	unset := repository.Exists(repository.Field(model.OrganizationProvisioningLeaseField), false)
	expired := repository.Query().Comparison(repository.Field(model.OrganizationProvisioningLeaseField), builder.Lte, now)
	mine := repository.Query().Filter(repository.Field(model.OrganizationProvisioningLeaseOwnerField), leaseOwner)
	filter := repository.Query().
		Filter(repository.IDField(), organizationRef).
		Filter(repository.Field(model.OrganizationSignupStateField), string(model.OrganizationSignupStatePending)).
		Comparison(repository.Field(model.OrganizationProvisioningOwnerRefField), builder.Exists, true).
		Or(unset, expired, mine)

	patch := repository.Patch().
		Set(repository.Field(model.OrganizationProvisioningLeaseOwnerField), leaseOwner).
		Set(repository.Field(model.OrganizationProvisioningLeaseField), expiresAt)

	matched, err := db.DBImp.PatchMany(ctx, filter, patch)
	if err != nil {
		db.DBImp.Logger.Warn("Failed to acquire organization provisioning lease",
			zap.Error(err), mzap.OrgRef(organizationRef), zap.String("lease_owner", leaseOwner))
		return false, err
	}
	return matched > 0, nil
}

// ReleaseProvisioningLease frees the lease so another attempt can take over
// immediately instead of waiting out the expiry. It only matches the lease of
// the given owner: a lease taken over by a newer attempt stays untouched.
func (db *OrganizationDB) ReleaseProvisioningLease(
	ctx context.Context,
	organizationRef bson.ObjectID,
	leaseOwner string,
) error {
	filter := repository.Query().
		Filter(repository.IDField(), organizationRef).
		Filter(repository.Field(model.OrganizationProvisioningLeaseOwnerField), leaseOwner)

	patch := repository.Patch().
		Unset(repository.Field(model.OrganizationProvisioningLeaseOwnerField)).
		Unset(repository.Field(model.OrganizationProvisioningLeaseField))

	if _, err := db.DBImp.PatchMany(ctx, filter, patch); err != nil {
		db.DBImp.Logger.Warn("Failed to release organization provisioning lease",
			zap.Error(err), mzap.OrgRef(organizationRef), zap.String("lease_owner", leaseOwner))
		return err
	}
	return nil
}

// FinalizeProvisioning atomically activates a pending organization while the
// given attempt still owns its provisioning lease. It reports false when the
// organization is no longer pending or the lease was taken over, so a slow
// attempt cannot finalize an organization another attempt now works on.
// Only the lease is cleared: the provisioning provenance stays on the document
// so a retry after a lost success response can recognize the activated
// organization instead of creating a duplicate.
func (db *OrganizationDB) FinalizeProvisioning(
	ctx context.Context,
	organizationRef bson.ObjectID,
	leaseOwner string,
) (bool, error) {
	filter := repository.Query().
		Filter(repository.IDField(), organizationRef).
		Filter(repository.Field(model.OrganizationSignupStateField), string(model.OrganizationSignupStatePending)).
		Filter(repository.Field(model.OrganizationProvisioningLeaseOwnerField), leaseOwner)

	patch := repository.Patch().
		Set(repository.Field(model.OrganizationSignupStateField), model.OrganizationSignupStateActive).
		Unset(repository.Field(model.OrganizationProvisioningLeaseOwnerField)).
		Unset(repository.Field(model.OrganizationProvisioningLeaseField))

	matched, err := db.DBImp.PatchMany(ctx, filter, patch)
	if err != nil {
		db.DBImp.Logger.Warn("Failed to finalize organization provisioning",
			zap.Error(err), mzap.OrgRef(organizationRef), zap.String("lease_owner", leaseOwner))
		return false, err
	}
	return matched > 0, nil
}

// FindPendingProvisioning returns the organization that a previous attempt left
// pending for the same owner and name, if any. The name matches the immutable
// provisioning snapshot, not the mutable display name, so a rename cannot
// detach the organization from the retries of the create that produced it.
func (db *OrganizationDB) FindPendingProvisioning(
	ctx context.Context,
	ownerRef bson.ObjectID,
	name string,
) (*model.Organization, error) {
	query := repository.Query().
		Filter(repository.Field(model.OrganizationProvisioningOwnerRefField), ownerRef).
		Filter(repository.Field(model.OrganizationSignupStateField), string(model.OrganizationSignupStatePending)).
		Filter(repository.Field(model.OrganizationProvisioningNameField), name)

	org := &model.Organization{}
	if err := db.DBImp.FindOne(ctx, query, org); err != nil {
		if errors.Is(err, merrors.ErrNoData) {
			return nil, nil
		}
		db.DBImp.Logger.Warn("Failed to look up pending organization provisioning",
			zap.Error(err), mzap.AccRef(ownerRef), zap.String("organization_name", name))
		return nil, err
	}
	return org, nil
}

// FindActiveProvisioned returns the already activated organization a previous
// attempt provisioned for the same owner and name, if any. It is the retry
// barrier for the success that got lost on the wire: without it a repeated
// create after a timed out but committed finalization would provision a
// duplicate organization. The name matches the immutable provisioning
// snapshot, so the barrier survives a rename of the activated organization.
func (db *OrganizationDB) FindActiveProvisioned(
	ctx context.Context,
	ownerRef bson.ObjectID,
	name string,
) (*model.Organization, error) {
	// Archived organizations release their name: they are excluded here and in
	// the unique index alike, so the checks and the constraint stay in step.
	query := repository.Query().
		Filter(repository.Field(model.OrganizationProvisioningOwnerRefField), ownerRef).
		Filter(repository.Field(model.OrganizationSignupStateField), string(model.OrganizationSignupStateActive)).
		Filter(repository.Field(model.OrganizationProvisioningNameField), name).
		Filter(repository.Field(storable.IsArchivedField), false)

	org := &model.Organization{}
	if err := db.DBImp.FindOne(ctx, query, org); err != nil {
		if errors.Is(err, merrors.ErrNoData) {
			return nil, nil
		}
		db.DBImp.Logger.Warn("Failed to look up activated organization provisioning",
			zap.Error(err), mzap.AccRef(ownerRef), zap.String("organization_name", name))
		return nil, err
	}
	return org, nil
}