Add optimistic membership lifecycle

2026-09-03 23:22:19 -04:00
parent fe6bd94c9a
commit 59827bf641
10 changed files with 368 additions and 20 deletions
@@ -453,6 +453,48 @@ func (store *Store) SetMembershipStatus(ctx context.Context, organizationID, use
 	return tx.Commit()
 }

+func (store *Store) ChangeMembershipStatus(ctx context.Context, input organizations.MembershipStatusChange, ownerRole string, audit organizations.AuditEvent) error {
+	if !validMembershipStatusChange(input, ownerRole, audit) {
+		return organizations.ErrMembershipNotFound
+	}
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if err = lockActiveMembershipActor(ctx, tx, input.OrganizationID, input.ActorUserID); err != nil {
+		return err
+	}
+	current, err := membershipStatus(ctx, tx, input.OrganizationID, input.UserID)
+	if err != nil {
+		return err
+	}
+	if current != input.ExpectedStatus {
+		return organizations.ErrRevisionConflict
+	}
+	if input.Status == "suspended" {
+		if err = protectLastOwner(ctx, tx, input.OrganizationID, input.UserID, ownerRole); err != nil {
+			return err
+		}
+	}
+	result, err := tx.ExecContext(ctx, `UPDATE gwf_organization_memberships SET status=? WHERE organization_id=? AND user_id=? AND status=?`, input.Status, input.OrganizationID, input.UserID, input.ExpectedStatus)
+	if err != nil {
+		return err
+	}
+	if changed, _ := result.RowsAffected(); changed != 1 {
+		return organizations.ErrRevisionConflict
+	}
+	if input.Status == "suspended" {
+		if _, err = tx.ExecContext(ctx, `DELETE FROM gwf_team_members WHERE user_id=? AND team_id IN (SELECT id FROM gwf_teams WHERE organization_id=?)`, input.UserID, input.OrganizationID); err != nil {
+			return err
+		}
+	}
+	if err = appendOrganizationAudit(ctx, tx, audit); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
 func (store *Store) RemoveMembership(ctx context.Context, organizationID, userID, ownerRole string, audit organizations.AuditEvent) error {
 	if !opaqueID(organizationID) || !opaqueID(userID) || !safeName(ownerRole) || !validOrganizationAudit(audit, organizationID) {
 		return organizations.ErrMembershipNotFound
@@ -484,6 +526,77 @@ func (store *Store) RemoveMembership(ctx context.Context, organizationID, userID
 	return tx.Commit()
 }

+func (store *Store) RemoveMembershipIfCurrent(ctx context.Context, input organizations.MembershipRemoval, ownerRole string, audit organizations.AuditEvent) error {
+	if !validMembershipRemoval(input, ownerRole, audit) {
+		return organizations.ErrMembershipNotFound
+	}
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if err = lockActiveMembershipActor(ctx, tx, input.OrganizationID, input.ActorUserID); err != nil {
+		return err
+	}
+	current, err := membershipStatus(ctx, tx, input.OrganizationID, input.UserID)
+	if err != nil {
+		return err
+	}
+	if current != input.ExpectedStatus {
+		return organizations.ErrRevisionConflict
+	}
+	if err = protectLastOwner(ctx, tx, input.OrganizationID, input.UserID, ownerRole); err != nil {
+		return err
+	}
+	if _, err = tx.ExecContext(ctx, `DELETE FROM gwf_team_members WHERE user_id=? AND team_id IN (SELECT id FROM gwf_teams WHERE organization_id=?)`, input.UserID, input.OrganizationID); err != nil {
+		return err
+	}
+	if _, err = tx.ExecContext(ctx, `UPDATE gwf_access_bindings SET revoked_by_user_id=?,revoked_at=? WHERE organization_id=? AND subject_kind='user' AND subject_id=? AND revoked_at IS NULL`, audit.ActorUserID, audit.CreatedAt.Unix(), input.OrganizationID, input.UserID); err != nil {
+		return err
+	}
+	result, err := tx.ExecContext(ctx, `DELETE FROM gwf_organization_memberships WHERE organization_id=? AND user_id=? AND status=?`, input.OrganizationID, input.UserID, input.ExpectedStatus)
+	if err != nil {
+		return err
+	}
+	if changed, _ := result.RowsAffected(); changed != 1 {
+		return organizations.ErrRevisionConflict
+	}
+	if err = appendOrganizationAudit(ctx, tx, audit); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+func lockActiveMembershipActor(ctx context.Context, tx *sql.Tx, organizationID, actorUserID string) error {
+	// Acquire the SQLite write lock before reading the optimistic state. This
+	// makes a competing lifecycle transaction observe the committed winner.
+	result, err := tx.ExecContext(ctx, `UPDATE gwf_organization_memberships SET status=status
+		WHERE organization_id=? AND user_id=? AND status='active'
+		AND EXISTS (SELECT 1 FROM gwf_organizations o WHERE o.id=? AND o.status='active')
+		AND EXISTS (SELECT 1 FROM gwf_users u WHERE u.id=? AND u.status='active')`, organizationID, actorUserID, organizationID, actorUserID)
+	if err != nil {
+		return err
+	}
+	if changed, _ := result.RowsAffected(); changed != 1 {
+		return organizations.ErrMembershipNotFound
+	}
+	return nil
+}
+
+func membershipStatus(ctx context.Context, tx *sql.Tx, organizationID, userID string) (string, error) {
+	var status string
+	if err := tx.QueryRowContext(ctx, `SELECT status FROM gwf_organization_memberships WHERE organization_id=? AND user_id=?`, organizationID, userID).Scan(&status); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return "", organizations.ErrMembershipNotFound
+		}
+		return "", err
+	}
+	if status != "active" && status != "suspended" {
+		return "", errors.New("authsqlite: stored membership status is invalid")
+	}
+	return status, nil
+}
+
 func protectLastOwner(ctx context.Context, tx *sql.Tx, organizationID, userID, ownerRole string) error {
 	var targetIsOwner int
 	if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM gwf_access_bindings WHERE organization_id=? AND subject_kind='user' AND subject_id=? AND role_name=? AND project_id IS NULL AND environment_id IS NULL AND service_id IS NULL AND revoked_at IS NULL`, organizationID, userID, ownerRole).Scan(&targetIsOwner); err != nil {
@@ -492,16 +605,37 @@ func protectLastOwner(ctx context.Context, tx *sql.Tx, organizationID, userID, o
 	if targetIsOwner == 0 {
 		return nil
 	}
-	var activeOwners int
-	if err := tx.QueryRowContext(ctx, `SELECT COUNT(DISTINCT b.subject_id) FROM gwf_access_bindings b JOIN gwf_organization_memberships m ON m.organization_id=b.organization_id AND m.user_id=b.subject_id AND m.status='active' WHERE b.organization_id=? AND b.subject_kind='user' AND b.role_name=? AND b.project_id IS NULL AND b.environment_id IS NULL AND b.service_id IS NULL AND b.revoked_at IS NULL`, organizationID, ownerRole).Scan(&activeOwners); err != nil {
+	var otherActiveOwners int
+	if err := tx.QueryRowContext(ctx, `SELECT COUNT(DISTINCT b.subject_id)
+		FROM gwf_access_bindings b
+		JOIN gwf_organization_memberships m ON m.organization_id=b.organization_id AND m.user_id=b.subject_id AND m.status='active'
+		JOIN gwf_users u ON u.id=m.user_id AND u.status='active'
+		WHERE b.organization_id=? AND b.subject_kind='user' AND b.subject_id<>? AND b.role_name=?
+		AND b.project_id IS NULL AND b.environment_id IS NULL AND b.service_id IS NULL
+		AND b.revoked_at IS NULL`, organizationID, userID, ownerRole).Scan(&otherActiveOwners); err != nil {
 		return err
 	}
-	if activeOwners <= 1 {
+	if otherActiveOwners == 0 {
 		return organizations.ErrLastOwner
 	}
 	return nil
 }

+func validMembershipStatusChange(input organizations.MembershipStatusChange, ownerRole string, audit organizations.AuditEvent) bool {
+	return opaqueID(input.OrganizationID) && opaqueID(input.UserID) && opaqueID(input.ActorUserID) && safeName(ownerRole) &&
+		(input.ExpectedStatus == "active" || input.ExpectedStatus == "suspended") &&
+		(input.Status == "active" || input.Status == "suspended") && input.ExpectedStatus != input.Status &&
+		validOrganizationAudit(audit, input.OrganizationID) && audit.ActorUserID == input.ActorUserID &&
+		audit.Action == "membership."+input.Status && audit.ResourceType == "membership" && audit.ResourceID == input.UserID && audit.RequestID == input.RequestID
+}
+
+func validMembershipRemoval(input organizations.MembershipRemoval, ownerRole string, audit organizations.AuditEvent) bool {
+	return opaqueID(input.OrganizationID) && opaqueID(input.UserID) && opaqueID(input.ActorUserID) && safeName(ownerRole) &&
+		(input.ExpectedStatus == "active" || input.ExpectedStatus == "suspended") &&
+		validOrganizationAudit(audit, input.OrganizationID) && audit.ActorUserID == input.ActorUserID &&
+		audit.Action == "membership.remove" && audit.ResourceType == "membership" && audit.ResourceID == input.UserID && audit.RequestID == input.RequestID
+}
+
 func (store *Store) Invitations(ctx context.Context, organizationID string, limit int) ([]organizations.Invitation, error) {
 	if !opaqueID(organizationID) || limit < 1 || limit > 1000 {
 		return nil, errors.New("authsqlite: invalid invitation query")
@@ -438,7 +438,7 @@ func TestInvitationAccessLifecycleAndLastOwnerProtection(t *testing.T) {
 	if _, err = accessService.Grant(t.Context(), access.Grant{SubjectKind: access.User, SubjectID: owner.ID, Role: "organization.owner", Scope: access.Scope{OrganizationID: organization.ID}, GrantedBy: owner.ID}); err != nil {
 		t.Fatal(err)
 	}
-	if err = organizationService.SetMembershipStatus(t.Context(), organization.ID, owner.ID, "suspended", owner.ID, "request-last-owner"); !errors.Is(err, organizations.ErrLastOwner) {
+	if err = organizationService.ChangeMembershipStatus(t.Context(), organizations.MembershipStatusChange{OrganizationID: organization.ID, UserID: owner.ID, ExpectedStatus: "active", Status: "suspended", ActorUserID: owner.ID, RequestID: "request-last-owner"}); !errors.Is(err, organizations.ErrLastOwner) {
 		t.Fatalf("last-owner suspension err=%v", err)
 	}
 	team, err := organizationService.CreateTeam(t.Context(), organizations.CreateTeam{OrganizationID: organization.ID, Slug: "operators", Name: "Operators", ActorUserID: owner.ID})
@@ -463,10 +463,10 @@ func TestInvitationAccessLifecycleAndLastOwnerProtection(t *testing.T) {
 	if err != nil || len(teams) != 1 || teams[0].ID != team.ID {
 		t.Fatalf("member teams=%+v err=%v", teams, err)
 	}
-	if err = organizationService.SetMembershipStatus(t.Context(), organization.ID, owner.ID, "suspended", owner.ID, "request-suspend-owner"); err != nil {
+	if err = organizationService.ChangeMembershipStatus(t.Context(), organizations.MembershipStatusChange{OrganizationID: organization.ID, UserID: owner.ID, ExpectedStatus: "active", Status: "suspended", ActorUserID: owner.ID, RequestID: "request-suspend-owner"}); err != nil {
 		t.Fatal(err)
 	}
-	if err = organizationService.RemoveMembership(t.Context(), organization.ID, member.ID, member.ID, "request-last-member"); !errors.Is(err, organizations.ErrLastOwner) {
+	if err = organizationService.RemoveMembershipIfCurrent(t.Context(), organizations.MembershipRemoval{OrganizationID: organization.ID, UserID: member.ID, ExpectedStatus: "active", ActorUserID: member.ID, RequestID: "request-last-member"}); !errors.Is(err, organizations.ErrLastOwner) {
 		t.Fatalf("sole active owner removal err=%v", err)
 	}
 	if _, err = organizationService.SetOrganizationStatus(t.Context(), organizations.SetOrganizationStatus{ID: organization.ID, Status: "archived", ActorUserID: member.ID, ExpectedRevision: organization.Revision, RequestID: "request-archive"}); err != nil {
@@ -635,6 +635,113 @@ func TestOrganizationRoleAdministrationIsAtomicAndProtectsOwners(t *testing.T) {
 	assertCount(t, store, `SELECT COUNT(*) FROM gwf_access_bindings WHERE id=?`, replacement.ID, 0)
 }

+func TestOptimisticMembershipLifecycleIsSerializedAndAtomic(t *testing.T) {
+	store, err := Open(filepath.Join(t.TempDir(), "accounts.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+	now := time.Date(2026, 9, 4, 9, 0, 0, 0, time.UTC)
+	authService, err := auth.New(store, auth.Options{Now: func() time.Time { return now }})
+	if err != nil {
+		t.Fatal(err)
+	}
+	owner, err := authService.CreateUser(t.Context(), auth.CreateUser{Username: "lifecycle.owner", Email: "lifecycle-owner@example.test", DisplayName: "Lifecycle Owner", Password: "correct horse battery staple"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	member, err := authService.CreateUser(t.Context(), auth.CreateUser{Username: "lifecycle.member", Email: "lifecycle-member@example.test", DisplayName: "Lifecycle Member", Password: "correct horse battery staple"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	organizationService, err := organizations.New(store, organizations.Options{Now: func() time.Time { return now }, OwnerRole: "owner"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	organization, err := organizationService.CreateOrganization(t.Context(), organizations.CreateOrganization{Slug: "optimistic-lifecycle", Name: "Optimistic Lifecycle", OwnerUserID: owner.ID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	policy := access.Policy{Roles: map[string]string{"owner": "Owner", "viewer": "Viewer"}, Permissions: map[string]string{"telemetry.read": "Read"}, Grants: map[string][]string{"owner": {"telemetry.read"}, "viewer": {"telemetry.read"}}}
+	accessService, err := access.New(store, policy, access.Options{Now: func() time.Time { return now }, OwnerRole: "owner"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = accessService.Seed(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	if _, err = accessService.Grant(t.Context(), access.Grant{SubjectKind: access.User, SubjectID: owner.ID, Role: "owner", Scope: access.Scope{OrganizationID: organization.ID}, GrantedBy: owner.ID}); err != nil {
+		t.Fatal(err)
+	}
+	team, err := organizationService.CreateTeam(t.Context(), organizations.CreateTeam{OrganizationID: organization.ID, Slug: "operators", Name: "Operators", ActorUserID: owner.ID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw, _, err := organizationService.InviteWithAccess(t.Context(), organizations.InviteWithAccess{OrganizationID: organization.ID, Email: member.Email, InvitedByUserID: owner.ID, DirectRole: "viewer", TeamIDs: []string{team.ID}, Lifetime: 24 * time.Hour})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = organizationService.AcceptInvitation(t.Context(), raw, member.ID); err != nil {
+		t.Fatal(err)
+	}
+
+	start := make(chan struct{})
+	results := make(chan error, 2)
+	for _, requestID := range []string{"request-suspend-one", "request-suspend-two"} {
+		requestID := requestID
+		go func() {
+			<-start
+			results <- organizationService.ChangeMembershipStatus(t.Context(), organizations.MembershipStatusChange{OrganizationID: organization.ID, UserID: member.ID, ExpectedStatus: "active", Status: "suspended", ActorUserID: owner.ID, RequestID: requestID})
+		}()
+	}
+	close(start)
+	var successful, conflicted int
+	for range 2 {
+		switch lifecycleErr := <-results; {
+		case lifecycleErr == nil:
+			successful++
+		case errors.Is(lifecycleErr, organizations.ErrRevisionConflict):
+			conflicted++
+		default:
+			t.Fatalf("concurrent membership suspension err=%v", lifecycleErr)
+		}
+	}
+	if successful != 1 || conflicted != 1 {
+		t.Fatalf("concurrent membership suspension success=%d conflict=%d", successful, conflicted)
+	}
+	assertCount(t, store, `SELECT COUNT(*) FROM gwf_access_audit_events WHERE action='membership.suspended' AND resource_id=?`, member.ID, 1)
+	assertCount(t, store, `SELECT COUNT(*) FROM gwf_team_members WHERE user_id=?`, member.ID, 0)
+	decision, err := accessService.Authorize(t.Context(), member.ID, access.Scope{OrganizationID: organization.ID}, "telemetry.read")
+	if err != nil || decision.Allowed {
+		t.Fatalf("suspended member decision=%+v err=%v", decision, err)
+	}
+	if err = organizationService.RemoveMembershipIfCurrent(t.Context(), organizations.MembershipRemoval{OrganizationID: organization.ID, UserID: member.ID, ExpectedStatus: "active", ActorUserID: owner.ID, RequestID: "request-stale-remove"}); !errors.Is(err, organizations.ErrRevisionConflict) {
+		t.Fatalf("stale membership removal err=%v", err)
+	}
+	assertCount(t, store, `SELECT COUNT(*) FROM gwf_access_audit_events WHERE request_id=?`, "request-stale-remove", 0)
+	assertCount(t, store, `SELECT COUNT(*) FROM gwf_organization_memberships WHERE user_id=?`, member.ID, 1)
+
+	if err = organizationService.ChangeMembershipStatus(t.Context(), organizations.MembershipStatusChange{OrganizationID: organization.ID, UserID: member.ID, ExpectedStatus: "suspended", Status: "active", ActorUserID: owner.ID, RequestID: "request-reactivate"}); err != nil {
+		t.Fatal(err)
+	}
+	assertCount(t, store, `SELECT COUNT(*) FROM gwf_team_members WHERE user_id=?`, member.ID, 0)
+	if err = organizationService.ChangeMembershipStatus(t.Context(), organizations.MembershipStatusChange{OrganizationID: organization.ID, UserID: member.ID, ExpectedStatus: "suspended", Status: "active", ActorUserID: owner.ID, RequestID: "request-stale-reactivate"}); !errors.Is(err, organizations.ErrRevisionConflict) {
+		t.Fatalf("stale membership reactivation err=%v", err)
+	}
+	assertCount(t, store, `SELECT COUNT(*) FROM gwf_access_audit_events WHERE request_id=?`, "request-stale-reactivate", 0)
+
+	if err = organizationService.RemoveMembershipIfCurrent(t.Context(), organizations.MembershipRemoval{OrganizationID: organization.ID, UserID: member.ID, ExpectedStatus: "active", ActorUserID: owner.ID, RequestID: "request-remove-member"}); err != nil {
+		t.Fatal(err)
+	}
+	assertCount(t, store, `SELECT COUNT(*) FROM gwf_organization_memberships WHERE user_id=?`, member.ID, 0)
+	assertCount(t, store, `SELECT COUNT(*) FROM gwf_access_bindings WHERE subject_id=? AND revoked_at IS NOT NULL`, member.ID, 1)
+	assertCount(t, store, `SELECT COUNT(*) FROM gwf_access_audit_events WHERE request_id=?`, "request-remove-member", 1)
+	decision, err = accessService.Authorize(t.Context(), member.ID, access.Scope{OrganizationID: organization.ID}, "telemetry.read")
+	if err != nil || decision.Allowed {
+		t.Fatalf("removed member decision=%+v err=%v", decision, err)
+	}
+}
+
 func membershipPresent(values []organizations.Membership, userID, status string) bool {
 	for _, value := range values {
 		if value.UserID == userID && value.Status == status {