Files
multica/server/internal/handler/workspace.go
Bohan Jiang f5c2994aed feat(workspace): revoke a member's runtimes when they leave or are removed (#2401)
* feat(workspace): revoke a member's runtimes when they leave or are removed

Previously, leaving or being removed from a workspace only deleted the
member row — every runtime the departed user owned in that workspace
remained in the DB, kept its daemon_token valid, and stayed reachable to
the workspace's other members. The departed user lost access but their
machine kept doing work.

This change converges the runtime state in the same transaction as the
member-row deletion: agents pinned to those runtimes are archived,
in-flight tasks are cancelled (so the daemon's per-task status poller
interrupts the running agent gracefully), the runtimes are forced
offline, and the daemon_token rows are deleted. After commit the
DaemonTokenCache is invalidated and agent:archived / daemon:register
events fire so connected clients reconcile immediately.

Server-side state convergence is the production safety net; the
daemon_token revoke takes effect once the mdt_ flow is live (today most
daemons fall back to PAT/JWT, and the member-row deletion is what stops
those requests via requireWorkspaceMember).

Daemon-side handling (recognising the resulting 401/404 and tearing down
the local pairing for that workspace) lands in a follow-up.

Co-authored-by: multica-agent <github@multica.ai>

* fix(workspace): also cancel tasks for archived agents on member revoke

CancelAgentTasksByRuntime only matched tasks whose runtime_id was in the
revoked set, missing a real path: agent.runtime_id can be reassigned via
UpdateAgent, but agent_task_queue.runtime_id keeps the value from when
the task was queued. So an agent currently bound to the leaving member's
runtime gets archived correctly, but its older tasks still pinned to a
prior runtime stay 'queued' — and ClaimAgentTask does not gate on
agent.archived_at, so those orphaned tasks remain claimable by the
prior runtime.

Replace CancelAgentTasksByRuntime with CancelAgentTasksByRuntimeOrAgent,
which OR-matches runtime_ids and the archived agent IDs in one UPDATE.
Pass the archived agent IDs through from revokeAndRemoveMember.

Adds TestDeleteMember_CancelsTasksFromAgentReassignment as a regression
guard: same agent, two runtimes, the older task on the surviving runtime
must end up cancelled while the surviving runtime stays online.

Co-authored-by: multica-agent <github@multica.ai>

---------

Co-authored-by: multica-agent <github@multica.ai>
2026-05-11 15:06:50 +08:00

668 lines
20 KiB
Go

package handler
import (
"encoding/json"
"log/slog"
"net/http"
"regexp"
"strings"
"github.com/go-chi/chi/v5"
"github.com/jackc/pgx/v5/pgtype"
"github.com/multica-ai/multica/server/internal/analytics"
"github.com/multica-ai/multica/server/internal/logger"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/pkg/protocol"
)
var nonAlpha = regexp.MustCompile(`[^a-zA-Z]`)
var workspaceSlugPattern = regexp.MustCompile(`^[a-z0-9]+(?:-[a-z0-9]+)*$`)
// generateIssuePrefix produces a 2-5 char uppercase prefix from a workspace name.
// Examples: "Jiayuan's Workspace" → "JIA", "My Team" → "MYT", "AB" → "AB".
func generateIssuePrefix(name string) string {
letters := nonAlpha.ReplaceAllString(name, "")
if len(letters) == 0 {
return "WS"
}
letters = strings.ToUpper(letters)
if len(letters) > 3 {
letters = letters[:3]
}
return letters
}
type WorkspaceResponse struct {
ID string `json:"id"`
Name string `json:"name"`
Slug string `json:"slug"`
Description *string `json:"description"`
Context *string `json:"context"`
Settings any `json:"settings"`
Repos any `json:"repos"`
IssuePrefix string `json:"issue_prefix"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
func workspaceToResponse(w db.Workspace) WorkspaceResponse {
var settings any
if w.Settings != nil {
json.Unmarshal(w.Settings, &settings)
}
if settings == nil {
settings = map[string]any{}
}
var repos any
if w.Repos != nil {
json.Unmarshal(w.Repos, &repos)
}
if repos == nil {
repos = []any{}
}
return WorkspaceResponse{
ID: uuidToString(w.ID),
Name: w.Name,
Slug: w.Slug,
Description: textToPtr(w.Description),
Context: textToPtr(w.Context),
Settings: settings,
Repos: repos,
IssuePrefix: w.IssuePrefix,
CreatedAt: timestampToString(w.CreatedAt),
UpdatedAt: timestampToString(w.UpdatedAt),
}
}
type MemberResponse struct {
ID string `json:"id"`
WorkspaceID string `json:"workspace_id"`
UserID string `json:"user_id"`
Role string `json:"role"`
CreatedAt string `json:"created_at"`
}
func memberToResponse(m db.Member) MemberResponse {
return MemberResponse{
ID: uuidToString(m.ID),
WorkspaceID: uuidToString(m.WorkspaceID),
UserID: uuidToString(m.UserID),
Role: m.Role,
CreatedAt: timestampToString(m.CreatedAt),
}
}
func (h *Handler) ListWorkspaces(w http.ResponseWriter, r *http.Request) {
userID, ok := requireUserID(w, r)
if !ok {
return
}
workspaces, err := h.Queries.ListWorkspaces(r.Context(), parseUUID(userID))
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to list workspaces")
return
}
resp := make([]WorkspaceResponse, len(workspaces))
for i, ws := range workspaces {
resp[i] = workspaceToResponse(ws)
}
writeJSON(w, http.StatusOK, resp)
}
func (h *Handler) GetWorkspace(w http.ResponseWriter, r *http.Request) {
id := workspaceIDFromURL(r, "id")
idUUID, ok := parseUUIDOrBadRequest(w, id, "workspace id")
if !ok {
return
}
ws, err := h.Queries.GetWorkspace(r.Context(), idUUID)
if err != nil {
writeError(w, http.StatusNotFound, "workspace not found")
return
}
writeJSON(w, http.StatusOK, workspaceToResponse(ws))
}
type CreateWorkspaceRequest struct {
Name string `json:"name"`
Slug string `json:"slug"`
Description *string `json:"description"`
Context *string `json:"context"`
IssuePrefix *string `json:"issue_prefix"`
}
func (h *Handler) CreateWorkspace(w http.ResponseWriter, r *http.Request) {
userID, ok := requireUserID(w, r)
if !ok {
return
}
var req CreateWorkspaceRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
req.Name = strings.TrimSpace(req.Name)
req.Slug = strings.ToLower(strings.TrimSpace(req.Slug))
if req.Name == "" || req.Slug == "" {
writeError(w, http.StatusBadRequest, "name and slug are required")
return
}
if !workspaceSlugPattern.MatchString(req.Slug) {
writeError(w, http.StatusBadRequest, "slug must contain only lowercase letters, numbers, and hyphens")
return
}
if isReservedSlug(req.Slug) {
writeError(w, http.StatusBadRequest, "slug is reserved")
return
}
tx, err := h.TxStarter.Begin(r.Context())
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to create workspace")
return
}
defer tx.Rollback(r.Context())
issuePrefix := generateIssuePrefix(req.Name)
if req.IssuePrefix != nil && strings.TrimSpace(*req.IssuePrefix) != "" {
issuePrefix = strings.ToUpper(strings.TrimSpace(*req.IssuePrefix))
}
qtx := h.Queries.WithTx(tx)
ws, err := qtx.CreateWorkspace(r.Context(), db.CreateWorkspaceParams{
Name: req.Name,
Slug: req.Slug,
Description: ptrToText(req.Description),
Context: ptrToText(req.Context),
IssuePrefix: issuePrefix,
})
if err != nil {
if isUniqueViolation(err) {
writeError(w, http.StatusConflict, "workspace slug already exists")
return
}
writeError(w, http.StatusInternalServerError, "failed to create workspace: "+err.Error())
return
}
_, err = qtx.CreateMember(r.Context(), db.CreateMemberParams{
WorkspaceID: ws.ID,
UserID: parseUUID(userID),
Role: "owner",
})
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to add owner: "+err.Error())
return
}
// Becoming a workspace member is the physical event that "completes" onboarding —
// keep this atomic with CreateMember so `member` and `onboarded_at`
// can never disagree. COALESCE in MarkUserOnboarded keeps it idempotent.
if _, err := qtx.MarkUserOnboarded(r.Context(), parseUUID(userID)); err != nil {
writeError(w, http.StatusInternalServerError, "failed to mark user onboarded")
return
}
if err := tx.Commit(r.Context()); err != nil {
writeError(w, http.StatusInternalServerError, "failed to create workspace")
return
}
// "Is this the user's first workspace?" is derived in PostHog by looking
// at whether they have a prior workspace_created event, not stamped at
// emit time. Stamping here would race under concurrent creates without
// a schema change, and the event stream answers the question exactly.
h.Analytics.Capture(analytics.WorkspaceCreated(userID, uuidToString(ws.ID)))
slog.Info("workspace created", append(logger.RequestAttrs(r), "workspace_id", uuidToString(ws.ID), "name", ws.Name, "slug", ws.Slug)...)
writeJSON(w, http.StatusCreated, workspaceToResponse(ws))
}
type UpdateWorkspaceRequest struct {
Name *string `json:"name"`
Description *string `json:"description"`
Context *string `json:"context"`
Settings any `json:"settings"`
Repos any `json:"repos"`
IssuePrefix *string `json:"issue_prefix"`
}
func (h *Handler) UpdateWorkspace(w http.ResponseWriter, r *http.Request) {
id := workspaceIDFromURL(r, "id")
idUUID, ok := parseUUIDOrBadRequest(w, id, "workspace id")
if !ok {
return
}
var req UpdateWorkspaceRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
params := db.UpdateWorkspaceParams{
ID: idUUID,
}
if req.Name != nil {
name := strings.TrimSpace(*req.Name)
if name == "" {
writeError(w, http.StatusBadRequest, "name is required")
return
}
params.Name = pgtype.Text{String: name, Valid: true}
}
if req.Description != nil {
params.Description = pgtype.Text{String: *req.Description, Valid: true}
}
if req.Context != nil {
params.Context = pgtype.Text{String: *req.Context, Valid: true}
}
if req.Settings != nil {
s, _ := json.Marshal(req.Settings)
params.Settings = s
}
if req.Repos != nil {
reposJSON, _ := json.Marshal(req.Repos)
params.Repos = reposJSON
}
if req.IssuePrefix != nil {
prefix := strings.ToUpper(strings.TrimSpace(*req.IssuePrefix))
if prefix != "" {
params.IssuePrefix = pgtype.Text{String: prefix, Valid: true}
}
}
ws, err := h.Queries.UpdateWorkspace(r.Context(), params)
if err != nil {
slog.Warn("update workspace failed", append(logger.RequestAttrs(r), "error", err, "workspace_id", id)...)
writeError(w, http.StatusInternalServerError, "failed to update workspace: "+err.Error())
return
}
slog.Info("workspace updated", append(logger.RequestAttrs(r), "workspace_id", id)...)
userID := requestUserID(r)
h.publish(protocol.EventWorkspaceUpdated, uuidToString(ws.ID), "member", userID, map[string]any{"workspace": workspaceToResponse(ws)})
writeJSON(w, http.StatusOK, workspaceToResponse(ws))
}
func (h *Handler) ListMembers(w http.ResponseWriter, r *http.Request) {
workspaceID := chi.URLParam(r, "id")
member, ok := h.requireWorkspaceMember(w, r, workspaceID, "workspace not found")
if !ok {
return
}
members, err := h.Queries.ListMembers(r.Context(), member.WorkspaceID)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to list members")
return
}
resp := make([]MemberResponse, len(members))
for i, m := range members {
resp[i] = memberToResponse(m)
}
writeJSON(w, http.StatusOK, resp)
}
type MemberWithUserResponse struct {
ID string `json:"id"`
WorkspaceID string `json:"workspace_id"`
UserID string `json:"user_id"`
Role string `json:"role"`
CreatedAt string `json:"created_at"`
Name string `json:"name"`
Email string `json:"email"`
AvatarURL *string `json:"avatar_url"`
}
func (h *Handler) ListMembersWithUser(w http.ResponseWriter, r *http.Request) {
workspaceID := workspaceIDFromURL(r, "id")
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
if !ok {
return
}
members, err := h.Queries.ListMembersWithUser(r.Context(), wsUUID)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to list members")
return
}
resp := make([]MemberWithUserResponse, len(members))
for i, m := range members {
resp[i] = MemberWithUserResponse{
ID: uuidToString(m.ID),
WorkspaceID: uuidToString(m.WorkspaceID),
UserID: uuidToString(m.UserID),
Role: m.Role,
CreatedAt: timestampToString(m.CreatedAt),
Name: m.UserName,
Email: m.UserEmail,
AvatarURL: textToPtr(m.UserAvatarUrl),
}
}
writeJSON(w, http.StatusOK, resp)
}
type CreateMemberRequest struct {
Email string `json:"email"`
Role string `json:"role"`
}
func memberWithUserResponse(member db.Member, user db.User) MemberWithUserResponse {
return MemberWithUserResponse{
ID: uuidToString(member.ID),
WorkspaceID: uuidToString(member.WorkspaceID),
UserID: uuidToString(member.UserID),
Role: member.Role,
CreatedAt: timestampToString(member.CreatedAt),
Name: user.Name,
Email: user.Email,
AvatarURL: textToPtr(user.AvatarUrl),
}
}
func normalizeMemberRole(role string) (string, bool) {
if role == "" {
return "member", true
}
role = strings.TrimSpace(role)
switch role {
case "owner", "admin", "member":
return role, true
default:
return "", false
}
}
func (h *Handler) CreateMember(w http.ResponseWriter, r *http.Request) {
workspaceID := workspaceIDFromURL(r, "id")
requester, ok := h.workspaceMember(w, r, workspaceID)
if !ok {
return
}
var req CreateMemberRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
email := strings.ToLower(strings.TrimSpace(req.Email))
if email == "" {
writeError(w, http.StatusBadRequest, "email is required")
return
}
role, valid := normalizeMemberRole(req.Role)
if !valid {
writeError(w, http.StatusBadRequest, "invalid member role")
return
}
if role == "owner" && requester.Role != "owner" {
writeError(w, http.StatusForbidden, "insufficient permissions")
return
}
user, err := h.Queries.GetUserByEmail(r.Context(), email)
if err != nil {
if isNotFound(err) {
// Auto-create user with email so they can be invited before signing up
user, err = h.Queries.CreateUser(r.Context(), db.CreateUserParams{
Name: email,
Email: email,
})
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to create user")
return
}
} else {
writeError(w, http.StatusInternalServerError, "failed to load user")
return
}
}
member, err := h.Queries.CreateMember(r.Context(), db.CreateMemberParams{
WorkspaceID: requester.WorkspaceID,
UserID: user.ID,
Role: role,
})
if err != nil {
if isUniqueViolation(err) {
writeError(w, http.StatusConflict, "user is already a member")
return
}
slog.Warn("create member failed", append(logger.RequestAttrs(r), "error", err, "workspace_id", workspaceID, "email", email)...)
writeError(w, http.StatusInternalServerError, "failed to create member")
return
}
slog.Info("member added", append(logger.RequestAttrs(r), "member_id", uuidToString(member.ID), "workspace_id", workspaceID, "email", email, "role", role)...)
userID := requestUserID(r)
eventPayload := map[string]any{"member": memberWithUserResponse(member, user)}
if ws, err := h.Queries.GetWorkspace(r.Context(), requester.WorkspaceID); err == nil {
eventPayload["workspace_name"] = ws.Name
}
h.publish(protocol.EventMemberAdded, uuidToString(requester.WorkspaceID), "member", userID, eventPayload)
writeJSON(w, http.StatusCreated, memberWithUserResponse(member, user))
}
type UpdateMemberRequest struct {
Role string `json:"role"`
}
func (h *Handler) UpdateMember(w http.ResponseWriter, r *http.Request) {
workspaceID := workspaceIDFromURL(r, "id")
requester, ok := h.workspaceMember(w, r, workspaceID)
if !ok {
return
}
memberID := chi.URLParam(r, "memberId")
memberUUID, ok := parseUUIDOrBadRequest(w, memberID, "member id")
if !ok {
return
}
target, err := h.Queries.GetMember(r.Context(), memberUUID)
if err != nil || uuidToString(target.WorkspaceID) != uuidToString(requester.WorkspaceID) {
writeError(w, http.StatusNotFound, "member not found")
return
}
var req UpdateMemberRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
if strings.TrimSpace(req.Role) == "" {
writeError(w, http.StatusBadRequest, "role is required")
return
}
role, valid := normalizeMemberRole(req.Role)
if !valid {
writeError(w, http.StatusBadRequest, "invalid member role")
return
}
if (target.Role == "owner" || role == "owner") && requester.Role != "owner" {
writeError(w, http.StatusForbidden, "insufficient permissions")
return
}
if target.Role == "owner" && role != "owner" {
members, err := h.Queries.ListMembers(r.Context(), target.WorkspaceID)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to update member")
return
}
if countOwners(members) <= 1 {
writeError(w, http.StatusBadRequest, "workspace must have at least one owner")
return
}
}
updatedMember, err := h.Queries.UpdateMemberRole(r.Context(), db.UpdateMemberRoleParams{
ID: target.ID,
Role: role,
})
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to update member")
return
}
user, err := h.Queries.GetUser(r.Context(), updatedMember.UserID)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to load member")
return
}
userID := requestUserID(r)
h.publish(protocol.EventMemberUpdated, uuidToString(requester.WorkspaceID), "member", userID, map[string]any{
"member": memberWithUserResponse(updatedMember, user),
})
writeJSON(w, http.StatusOK, memberWithUserResponse(updatedMember, user))
}
func (h *Handler) DeleteMember(w http.ResponseWriter, r *http.Request) {
workspaceID := workspaceIDFromURL(r, "id")
requester, ok := h.workspaceMember(w, r, workspaceID)
if !ok {
return
}
memberID := chi.URLParam(r, "memberId")
memberUUID, ok := parseUUIDOrBadRequest(w, memberID, "member id")
if !ok {
return
}
target, err := h.Queries.GetMember(r.Context(), memberUUID)
if err != nil || uuidToString(target.WorkspaceID) != uuidToString(requester.WorkspaceID) {
writeError(w, http.StatusNotFound, "member not found")
return
}
if target.Role == "owner" && requester.Role != "owner" {
writeError(w, http.StatusForbidden, "insufficient permissions")
return
}
if target.Role == "owner" {
members, err := h.Queries.ListMembers(r.Context(), target.WorkspaceID)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to delete member")
return
}
if countOwners(members) <= 1 {
writeError(w, http.StatusBadRequest, "workspace must have at least one owner")
return
}
}
requesterUserID := requestUserID(r)
result, err := h.revokeAndRemoveMember(r.Context(), target.WorkspaceID, target.UserID, target.ID, parseUUID(requesterUserID))
if err != nil {
slog.Warn("delete member failed", append(logger.RequestAttrs(r), "error", err, "member_id", memberID, "workspace_id", workspaceID)...)
writeError(w, http.StatusInternalServerError, "failed to delete member")
return
}
wsIDStr := uuidToString(requester.WorkspaceID)
logRevocation(result, wsIDStr, uuidToString(target.UserID))
h.publishRevocation(r.Context(), result, wsIDStr, "member", requesterUserID)
slog.Info("member removed", append(logger.RequestAttrs(r), "member_id", uuidToString(target.ID), "workspace_id", workspaceID, "user_id", uuidToString(target.UserID))...)
h.publish(protocol.EventMemberRemoved, wsIDStr, "member", requesterUserID, map[string]any{
"member_id": uuidToString(target.ID),
"workspace_id": wsIDStr,
"user_id": uuidToString(target.UserID),
})
w.WriteHeader(http.StatusNoContent)
}
func (h *Handler) LeaveWorkspace(w http.ResponseWriter, r *http.Request) {
workspaceID := workspaceIDFromURL(r, "id")
member, ok := h.workspaceMember(w, r, workspaceID)
if !ok {
return
}
if member.Role == "owner" {
members, err := h.Queries.ListMembers(r.Context(), member.WorkspaceID)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to leave workspace")
return
}
if countOwners(members) <= 1 {
writeError(w, http.StatusBadRequest, "workspace must have at least one owner")
return
}
}
result, err := h.revokeAndRemoveMember(r.Context(), member.WorkspaceID, member.UserID, member.ID, member.UserID)
if err != nil {
slog.Warn("leave workspace failed", append(logger.RequestAttrs(r), "error", err, "workspace_id", workspaceID)...)
writeError(w, http.StatusInternalServerError, "failed to leave workspace")
return
}
userID := requestUserID(r)
logRevocation(result, workspaceID, uuidToString(member.UserID))
h.publishRevocation(r.Context(), result, workspaceID, "member", userID)
slog.Info("member removed", append(logger.RequestAttrs(r), "member_id", uuidToString(member.ID), "workspace_id", workspaceID, "user_id", uuidToString(member.UserID))...)
h.publish(protocol.EventMemberRemoved, workspaceID, "member", userID, map[string]any{
"member_id": uuidToString(member.ID),
"workspace_id": workspaceID,
"user_id": uuidToString(member.UserID),
})
w.WriteHeader(http.StatusNoContent)
}
func (h *Handler) DeleteWorkspace(w http.ResponseWriter, r *http.Request) {
workspaceID := workspaceIDFromURL(r, "id")
// Defense in depth: the route is already gated by the
// RequireWorkspaceRoleFromURL("owner") middleware, but we re-check here
// so that the handler is safe regardless of how it gets wired up
// (direct calls in tests, future router refactors, etc.).
requester, ok := h.workspaceMember(w, r, workspaceID)
if !ok {
return
}
if requester.Role != "owner" {
writeError(w, http.StatusForbidden, "insufficient permissions")
return
}
// At this point workspaceMember has resolved → workspaceID is a valid UUID
// (the lookup would have errored otherwise), so reuse the resolved value.
if err := h.Queries.DeleteWorkspace(r.Context(), requester.WorkspaceID); err != nil {
slog.Warn("delete workspace failed", append(logger.RequestAttrs(r), "error", err, "workspace_id", workspaceID)...)
writeError(w, http.StatusInternalServerError, "failed to delete workspace")
return
}
slog.Info("workspace deleted", append(logger.RequestAttrs(r), "workspace_id", workspaceID)...)
h.publish(protocol.EventWorkspaceDeleted, workspaceID, "member", requestUserID(r), map[string]any{
"workspace_id": workspaceID,
})
w.WriteHeader(http.StatusNoContent)
}