Files
2083970b51 MUL-8374: fix: keep private chat content to the session owner (realtime + REST) (#9677)
* MUL-8374: fix(realtime): deliver private chat content only to the session owner

Chat events and chat-task transcript (task:message) frames were fanned out
to every connection in the workspace. Route every chat:* event, and
task:message for chat tasks, to the chat session creator's user scope
instead. Producers stamp the owner on the bus event (from the same
chat_session row they already read); an event without an owner is dropped
rather than widened. Chat-task lifecycle events keep workspace fanout for
agent presence.

Co-authored-by: Cursor <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(server): keep chat-task content private on REST task reads

The transcript endpoint now 404s chat tasks for anyone but the chat
session's creator, and the agent run list and workspace task snapshot keep
foreign chat-task rows but blank their result and error.

Co-authored-by: Cursor <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(realtime): scope private chat delivery to the owner's workspace connections

User-scope delivery reached the chat creator's connections to every
workspace. Those are authorized against their own workspace only, so a
creator removed from the chat's workspace kept receiving its transcript
and reply through a socket to another workspace. Chat content now goes to
a workspace_member scope each connection joins at connect time; it is
re-authorized against the chat's workspace on every write.

Co-authored-by: Cursor <[email protected]>
Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: Cursor <[email protected]>
Co-authored-by: multica-agent <[email protected]>
2026-10-29 19:13:45 +08:00

1277 lines
53 KiB
Go

package handler
import (
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"log/slog"
"math"
"net/http"
"net/netip"
"strconv"
"strings"
"time"
"github.com/go-chi/chi/v5"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgtype"
"github.com/multica-ai/multica/server/internal/analytics"
"github.com/multica-ai/multica/server/internal/auth"
"github.com/multica-ai/multica/server/internal/cloudruntime"
"github.com/multica-ai/multica/server/internal/daemonws"
"github.com/multica-ai/multica/server/internal/dbreader"
"github.com/multica-ai/multica/server/internal/entitlement"
"github.com/multica-ai/multica/server/internal/events"
"github.com/multica-ai/multica/server/internal/integrations/channel/engine"
composio "github.com/multica-ai/multica/server/internal/integrations/composio"
"github.com/multica-ai/multica/server/internal/integrations/dingtalk"
"github.com/multica-ai/multica/server/internal/integrations/ghsnapshot"
"github.com/multica-ai/multica/server/internal/integrations/lark"
"github.com/multica-ai/multica/server/internal/integrations/slack"
"github.com/multica-ai/multica/server/internal/integrations/telegram"
"github.com/multica-ai/multica/server/internal/integrations/wecom"
"github.com/multica-ai/multica/server/internal/issuestatus"
obsmetrics "github.com/multica-ai/multica/server/internal/metrics"
"github.com/multica-ai/multica/server/internal/middleware"
"github.com/multica-ai/multica/server/internal/realtime"
"github.com/multica-ai/multica/server/internal/seatcapacity"
"github.com/multica-ai/multica/server/internal/service"
"github.com/multica-ai/multica/server/internal/storage"
"github.com/multica-ai/multica/server/internal/util"
"github.com/multica-ai/multica/server/internal/util/secretbox"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/pkg/featureflag"
"github.com/multica-ai/multica/server/pkg/llm"
)
// randomID returns a random 16-byte hex string used as a request ID for
// in-memory stores (model list, local skills, CLI update, etc.).
func randomID() string {
b := make([]byte, 16)
rand.Read(b)
return hex.EncodeToString(b)
}
type txStarter interface {
Begin(ctx context.Context) (pgx.Tx, error)
}
type dbExecutor interface {
Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}
type Config struct {
AllowSignup bool
AllowedEmails []string
AllowedEmailDomains []string
// DisableWorkspaceCreation, when true, makes POST /api/workspaces return
// 403 for every caller. There is no role/owner exception because the repo
// has no platform-admin concept; operators bootstrap the workspace with
// the flag off, then flip it on and restart so subsequent users join via
// invitation only. The public /api/config endpoint mirrors this flag so
// the UI can hide every "Create workspace" affordance — see #3433.
DisableWorkspaceCreation bool
// WorkspaceCreationAllowedEmails, when non-empty, restricts
// POST /api/workspaces to callers whose account email matches one of
// the entries (case-insensitive). Empty means every signed-in user may
// create workspaces. DisableWorkspaceCreation takes precedence. This only
// decides who may create a workspace; it grants no cross-workspace
// visibility or admin rights (#9400).
WorkspaceCreationAllowedEmails []string
// VCSIntegrationEnabled gates the self-hosted Git provider integration
// (Forgejo / Gitea / GitLab) at the deployment level, independent of whether
// MULTICA_VCS_SECRET_KEY is set. It is the product boundary: the feature is
// intended for self-hosted Multica only (where Multica and the Git instance
// can share a network), and is left off on the managed cloud — connect,
// rotate, and webhook handlers reject when it is false, and /api/config
// omits it so the UI hides the whole section rather than showing a
// "missing key" message a cloud user cannot act on. Populated from
// MULTICA_VCS_INTEGRATION_ENABLED; the self-host compose defaults it on.
VCSIntegrationEnabled bool
// PublicURL is the absolute base URL the API is reachable at from the
// public internet, with no trailing slash (e.g. "https://multica.ai").
// Used to build webhook_url responses and the fixed Remote MCP OAuth
// callback URI — never to decide request identity, routing, or workspace
// scope. Empty when unset; webhook clients can fall back to their own origin,
// while OAuth Connect fails closed because providers require an exact URI.
// Reading the public host from request headers (Host / X-Forwarded-Host)
// is intentionally avoided so a misconfigured reverse proxy cannot trick
// the server into minting webhook URLs pointing at an attacker-controlled
// host.
PublicURL string
// AppURL is the browser application's canonical origin, resolved from
// MULTICA_APP_URL (falling back to FRONTEND_ORIGIN). It is kept separate
// from PublicURL because split app/API deployments use different origins.
AppURL string
// TrustedProxies are CIDRs whose source IP we trust to set
// X-Forwarded-For / X-Real-IP. Empty means "trust nothing": the rate
// limiter uses r.RemoteAddr exclusively. Populated via the
// MULTICA_TRUSTED_PROXIES env var (comma-separated CIDRs, e.g.
// "10.0.0.0/8,127.0.0.1/32"). This is specifically to keep the per-IP
// webhook limiter from being bypassed by a spoofed XFF on deployments
// without a header-stripping reverse proxy in front.
TrustedProxies []netip.Prefix
// CloudURL enables the SaaS-only multica-cloud connection when set. Empty
// keeps self-hosted deployments explicit: Cloud endpoints return 403 instead
// of attempting to dial a hard-coded private service.
CloudURL string
CloudTimeout time.Duration
AttachmentDownloadMode string
AttachmentDownloadURLTTL time.Duration
// AttachmentFrameAncestors are trusted browser origins allowed to embed
// attachment preview responses. In production this should mirror the
// frontend/CORS origin allowlist so split app/api self-hosted deployments
// can frame API-hosted PDFs without allowing arbitrary third-party frames.
AttachmentFrameAncestors []string
// PluginSurfaceOrigin is the dedicated, cookie-free browser origin that
// routes /plugin-surfaces/* back to this server. It must not be the app/API
// origin: the route serves stored third-party JavaScript as HTML.
PluginSurfaceOrigin string
// LLM* configure the basic LLM API layer (MUL-4238). They back the
// server-internal LLM helpers in pkg/llm (e.g. chat title generation).
// The generic OpenAI-compatible passthrough endpoints were removed in
// MUL-4309; LLM access is internal-only now. When both LLMAPIKey and
// LLMBaseURL are empty the layer is disabled and callers fall back
// silently (see maybeGenerateChatTitleAsync).
// - LLMAPIKey -> MULTICA_LLM_API_KEY
// - LLMBaseURL -> MULTICA_LLM_BASE_URL (OpenAI or any compatible gateway)
// - LLMDefaultModel -> MULTICA_LLM_DEFAULT_MODEL (used when a request omits `model`)
// - LLMMaxRetries -> MULTICA_LLM_MAX_RETRIES (transport retry budget)
// - LLMDisableThinking -> MULTICA_LLM_DISABLE_THINKING (gateway hint to
// turn model reasoning off; see llm.Config.DisableThinking)
LLMAPIKey string
LLMBaseURL string
LLMDefaultModel string
// LLMMaxRetries is the parsed MULTICA_LLM_MAX_RETRIES budget. nil means
// unset (llm.DefaultMaxRetries applies); llm.Retries(0) disables retries.
// The type carries the validation: it can only be built through llm.Retries,
// and cmd/server additionally fails the boot on an out-of-range value before
// one reaches this struct. See llm.Config.MaxRetries for the full semantics.
LLMMaxRetries *llm.RetryOverride
// LLMDisableThinking is the parsed MULTICA_LLM_DISABLE_THINKING switch.
// cmd/server validates the raw value before the boot continues, so a
// non-boolean never reaches this struct.
LLMDisableThinking bool
// ServerVersion is the build version of the running API binary (the same
// value main.go stamps via -X main.version and reports on /metrics).
// Surfaced through /api/config so self-hosted operators can confirm which
// server build is deployed. Empty in dev builds.
ServerVersion string
// GitExtraCACerts is the PEM of the private CAs the operator added to the
// backend's trust store through SSL_CERT_DIR (see internal/gitca). It is
// handed to daemons with the workspace repos so their git can reach a
// repository host the backend already trusts. Empty when none are set.
GitExtraCACerts []byte
}
type cloudRuntimeProxy interface {
Enabled() bool
Do(ctx context.Context, req cloudruntime.Request) (*cloudruntime.Response, error)
}
type RuntimeProfileRefreshNotifier interface {
NotifyRuntimeProfilesChanged(workspaceID, profileID string)
}
type WorkspaceSetRefreshNotifier interface {
NotifyWorkspacesChanged(userID string)
}
// DaemonPendingWorkNotifier pushes a runtime-scoped "heartbeat now" hint to the
// daemon so a queued heartbeat-carried request (model discovery, capability
// discovery, or local-skill import) is picked up immediately instead of on the
// daemon's next scheduled tick (MUL-5444).
// Satisfied by both *daemonws.Hub (single-node) and *daemonws.RelayNotifier
// (multi-node, fans out through Redis).
type DaemonPendingWorkNotifier interface {
NotifyPendingWork(runtimeID, kind string)
}
// DaemonTaskSupplementNotifier sends a content-free wakeup for one exact run.
// The daemon still pulls and authenticates the durable supplement over HTTP.
type DaemonTaskSupplementNotifier interface {
NotifyTaskSupplementAvailable(runtimeID, taskID string)
}
// RuntimeGoneNotifier invalidates a runtime that was deleted while its daemon
// still has an authenticated WebSocket connection.
type RuntimeGoneNotifier interface {
NotifyRuntimeGone(runtimeID string)
}
// RuntimeRecoveryNotifier republishes the daemon:register lifecycle refresh
// after a heartbeat actually flipped an offline runtime row back online
// (sweeper-race fallback, batch-receipt reconciliation). Recovery paths already
// know the workspace, so publishing never needs a follow-up database lookup.
type RuntimeRecoveryNotifier interface {
NotifyRuntimeRecovered(ctx context.Context, workspaceID string)
}
type Handler struct {
Queries *db.Queries
ReadSelector *dbreader.Selector
DB dbExecutor
TxStarter txStarter
Hub *realtime.Hub
DaemonHub *daemonws.Hub
DaemonProfileRefresh RuntimeProfileRefreshNotifier
DaemonWorkspaceRefresh WorkspaceSetRefreshNotifier
DaemonRuntimeGone RuntimeGoneNotifier
Bus *events.Bus
TaskService *service.TaskService
PluginService *service.PluginService
IssueService *service.IssueService
AutopilotService *service.AutopilotService
// Entitlements supplies workspace-scoped commercial gates. A nil provider
// preserves self-hosted behavior without extra reads.
Entitlements entitlement.Provider
// SeatCapacity executes Cloud's pre-purchased human-seat protocol. Nil or
// disabled preserves self-hosted behavior.
SeatCapacity seatcapacity.Executor
SeatCapacityLocker seatcapacity.WorkspaceLocker
SeatCapacityWorker *seatcapacity.Worker
EmailService *service.EmailService
UpdateStore UpdateStore
ModelListStore ModelListStore
LocalSkillListStore LocalSkillListStore
LocalSkillImportStore LocalSkillImportStore
FeatureFlags *featureflag.Service
// IssueStatusCatalog reads the workspace status catalog. Defaults to
// Queries; a test can substitute a counting wrapper to assert HOW MANY
// catalog reads a request performs, which is the only property that
// distinguishes the current one-read derivation from the N+1 it replaced.
// (MUL-6243)
IssueStatusCatalog issuestatus.Querier
LivenessStore LivenessStore
HeartbeatScheduler HeartbeatScheduler
Storage storage.Storage
CFSigner *auth.CloudFrontSigner
Analytics analytics.Client
// DaemonPendingWork pushes "heartbeat now" hints for queued
// heartbeat-carried requests (MUL-5444). Optional: when nil,
// requestDaemonPendingWork falls back to the local DaemonHub, which is the
// correct delivery scope for a single-node deployment.
DaemonPendingWork DaemonPendingWorkNotifier
DaemonTaskSupplement DaemonTaskSupplementNotifier
// ModelCatalogCache serves the last known good model list for a runtime so
// the picker can render without waiting for a daemon round trip
// (stale-while-revalidate, MUL-5444). Nil-safe: every call site treats a nil
// cache as a permanent miss and falls back to the full discovery flow.
ModelCatalogCache ModelCatalogCache
// Metrics is the shared business-metrics collector built by main.go.
// May be nil in tests / self-hosted with the metrics listener disabled;
// every Record* method is nil-safe and obsmetrics.RecordEvent treats a
// nil Metrics as "PostHog only".
Metrics *obsmetrics.BusinessMetrics
PATCache *auth.PATCache
DaemonTokenCache *auth.DaemonTokenCache
MembershipCache *auth.MembershipCache
QoderRateLimiter WebhookRateLimiter
WebhookRateLimiter WebhookRateLimiter
WebhookIPRateLimiter WebhookRateLimiter
WebhookAbsoluteIPRateLimiter WebhookRateLimiter
InvitationRateLimiters InvitationRateLimiters
WebhookDeliveryWorker *WebhookDeliveryWorker
CloudRuntime cloudRuntimeProxy
// Test-only HTTP override; nil uses the default client in production.
googleOAuthHTTPClient *http.Client
// Lark integration. All three are nil when the Lark master key
// (MULTICA_LARK_SECRET_KEY) is unset; the corresponding HTTP
// handlers return 403 in that case so a misconfigured self-host
// deployment surfaces a clear error instead of silently using a
// zero key. Wired in cmd/server/router.go after handler.New.
LarkInstallations *lark.InstallationService
LarkBindingTokens *lark.BindingTokenService
// LarkRegistration owns the device-flow install lifecycle: begin
// a registration session against accounts.feishu.cn, poll, and
// on success write lark_installation + the installer's
// lark_user_binding in one DB transaction. Nil when the at-rest
// key is unset or the RegistrationService failed to construct at
// boot.
LarkRegistration *lark.RegistrationService
// LarkAPIClient is the live transport that backs SendInteractiveCard,
// PatchInteractiveCard, SendBindingPromptCard, GetBotInfo. The
// router wires the real Lark HTTP client whenever
// MULTICA_LARK_SECRET_KEY is set; tests that need a no-op
// behaviour can swap in `lark.NewStubAPIClient(...)` directly. The
// UI consults IsConfigured() to decide whether to surface install
// entry points.
LarkAPIClient lark.APIClient
// Composio integration (MUL-3720). Nil when COMPOSIO_API_KEY is unset;
// the composio HTTP handlers return 403 in that case. Wired in
// cmd/server/router.go after handler.New.
Composio *composio.Service
// ChannelSupervisor owns the per-installation supervisor goroutines
// that hold the §4.4 WS lease and drive each channel.Channel
// (MUL-3620 generalized the Feishu-only Hub into this channel-agnostic
// engine). The router constructs it independently of platform secrets — it
// drives any channel type, not just Feishu. It remains nil when lease
// configuration is unsafe or a selected Redis backend fails its startup
// readiness check; each platform registers its Factory only when configured
// (Feishu when MULTICA_LARK_SECRET_KEY is set). The router does NOT
// call Run; the process owner (main.go) starts it under a long-running
// context and joins via WaitWithTimeout (bounded, fenced by
// ShutdownTimeout) during graceful shutdown so the lease renewer yields
// cleanly when the DB is healthy without blocking process exit if the
// pool is frozen — at worst the next replica waits the full TTL.
ChannelSupervisor *engine.Supervisor
// ChannelRouter is the channel-agnostic inbound pipeline (the shared
// handler the Supervisor injects into every Channel). main.go calls
// Drain on it during shutdown, after the Supervisor has stopped
// delivering events, to flush debounced run triggers and join in-flight
// reply goroutines. Built unconditionally (even without Lark).
ChannelRouter *engine.Router
// ChannelMediaReconciler settles the channel-media intent ledger
// (uploaded-but-unbound object reclaim). Built in cmd/server/router.go
// where the storage backend exists; main.go starts it as an independent
// worker goroutine. Nil when no storage backend is configured.
ChannelMediaReconciler *service.ChannelMediaReconciler
// LarkTyping owns the durable reaction cleanup worker, started under sweepCtx.
LarkTyping *lark.TypingIndicatorManager
// SlackInstall owns the bring-your-own-app Slack install lifecycle (register
// pasted tokens / list / revoke) and the at-rest encryption of each app's bot
// + app tokens (MUL-3666). Nil unless MULTICA_SLACK_SECRET_KEY is set.
SlackInstall *slack.InstallService
// SlackBindingTokens mints/redeems the user-binding tokens behind the
// "link your Slack account" prompt (MUL-3666). Nil unless Slack is
// configured (MULTICA_SLACK_SECRET_KEY set).
SlackBindingTokens *slack.BindingTokenService
// DingTalkInstall owns the bring-your-own-app DingTalk lifecycle. It is nil
// unless MULTICA_DINGTALK_SECRET_KEY is configured.
DingTalkInstall *dingtalk.InstallService
// DingTalkBindingTokens mints and redeems the single-use account-link tokens.
DingTalkBindingTokens *dingtalk.BindingTokenService
// SlackHistory backs the agent-facing `multica chat history` command: it
// reads a chat session's bound Slack conversation on demand (MUL-3871). Nil
// unless Slack is configured; GetChatChannelHistory then reports "no channel
// integration". A future platform satisfies the same reader interface.
SlackHistory ChatChannelHistoryReader
// WecomStore is the read/write handle over channel_installation rows scoped
// to channel_type='wecom'. Nil disables the wecom Web-UI endpoints (they
// return 403) and prevents boot from wiring the smart-bot supervisor.
WecomStore *wecom.Store
// WecomCredentials unseals a wecom installation's smart-bot secret for the
// WebSocket subscribe frame. Nil disables the wecom integration.
WecomCredentials wecom.CredentialsResolver
// WecomBindingTokens mints/redeems the user-binding tokens behind the
// "link your Multica account" prompt sent to first-time WeCom users
// (their aibot userid is a "T"-prefixed anonymized id with no relation
// to their real userid or email, so an explicit binding is required —
// see wecom/binding.go). Nil disables the redeem endpoint (returns 403)
// and the OutboundReplier's binding-prompt path.
WecomBindingTokens WecomBindingRedeemer
// WecomCredentialProbe overrides the install-time control check. Nil in
// production, which gets the real handshake probe; tests inject a fake so
// the install path runs without a socket.
WecomCredentialProbe wecom.CredentialProbe
// TelegramInstall owns the Telegram bot install lifecycle (register a
// pasted BotFather token / list / revoke) and the at-rest encryption of
// each bot's token. Nil unless MULTICA_TELEGRAM_SECRET_KEY is set.
TelegramInstall *telegram.InstallService
// TelegramBindingTokens mints/redeems the user-binding tokens behind the
// "link your Telegram account" prompt. Nil unless Telegram is configured.
TelegramBindingTokens *telegram.BindingTokenService
// TelegramOutbound owns the asynchronous terminal-delivery worker pool.
// The process owner starts and joins it; the synchronous event bus only
// enqueues EventChatDone work.
TelegramOutbound *telegram.Outbound
// channelFileDelivery names the channel types that can, IN THIS
// DEPLOYMENT, carry a file the agent produced the last hop into the
// conversation. It answers the claim response's
// chat_channel_delivers_files, which the agent's PER-TURN prompt turns into
// either "run `multica attachment upload`" or "describe the file in words"
// (daemon/prompt.go). Not the brief: the brief is the prompt cache prefix and
// this is a per-turn verdict, so stating it there made one session render two
// briefs (MUL-5377).
//
// It is a deployment fact, not a property of the channel type, and that
// distinction is the whole reason it lives here. Whether the file arrives
// takes TWO things: an adapter that goes back for the bound attachment,
// and object storage for it to go back to. The first is a property of the
// code, the second of the configuration, and only the process that wired
// both knows the conjunction. A daemon that answers this from the channel
// type alone promises delivery a storage-less deployment cannot perform.
//
// Written once at boot by DeclareChannelFileDelivery (cmd/server/router.go,
// in the same branch that passes the storage to the adapter, so the two
// cannot drift), read-only from then on. Nil means no channel delivers
// files, which is what a deployment with no storage configured gets.
channelFileDelivery map[string]bool
// LLM is the basic LLM API layer (MUL-4238): a thin wrapper over the
// OpenAI Go SDK backing server-internal one-shot LLM helpers such as chat
// title generation. The generic passthrough endpoints were removed in
// MUL-4309, so it is internal-only now. Always non-nil (New builds it from
// Config); when unconfigured its Enabled() reports false and callers fall
// back silently.
LLM *llm.Client
// VCSSecretBox encrypts/decrypts per-workspace Git provider access tokens and
// webhook secrets at rest (Forgejo / Gitea / GitLab). Nil when
// MULTICA_VCS_SECRET_KEY is unset; connect returns 403 and webhook returns 404
// in that case so a misconfigured self-host deployment surfaces a clear
// error rather than silently storing plaintext. Wired in
// cmd/server/router.go after New.
VCSSecretBox *secretbox.Box
// PluginSurfaceTokens seal short-lived launch claims. Nil disables surface
// launches; wired from a domain-separated MULTICA_PLUGIN_SECRET_KEY at boot.
PluginSurfaceTokens *secretbox.Box
// PRRefresh drives the GitHub API snapshot pipeline for PR cards (MUL-5265):
// webhook / page-visit / TTL triggers → authenticated GraphQL fetch →
// head-SHA-guarded atomic snapshot write. Always non-nil, but inert (every
// trigger is a no-op) when GITHUB_APP_ID / GITHUB_APP_PRIVATE_KEY are unset,
// so the feature degrades cleanly on deployments without a private key.
// Wired in cmd/server/router.go after New.
PRRefresh *ghsnapshot.Manager
cfg Config
}
func New(queries *db.Queries, txStarter txStarter, hub *realtime.Hub, bus *events.Bus, emailService *service.EmailService, store storage.Storage, cfSigner *auth.CloudFrontSigner, analyticsClient analytics.Client, cfg Config, daemonHubs ...*daemonws.Hub) *Handler {
var executor dbExecutor
if candidate, ok := txStarter.(dbExecutor); ok {
executor = candidate
}
if analyticsClient == nil {
analyticsClient = analytics.NoopClient{}
}
if mode, ok := normalizeAttachmentDownloadMode(cfg.AttachmentDownloadMode); ok {
cfg.AttachmentDownloadMode = string(mode)
} else {
slog.Warn("invalid ATTACHMENT_DOWNLOAD_MODE, using auto", "value", cfg.AttachmentDownloadMode)
cfg.AttachmentDownloadMode = string(attachmentDownloadModeAuto)
}
if cfg.AttachmentDownloadURLTTL <= 0 {
cfg.AttachmentDownloadURLTTL = defaultAttachmentDownloadURLTTL
}
var daemonHub *daemonws.Hub
if len(daemonHubs) > 0 {
daemonHub = daemonHubs[0]
}
var daemonProfileRefresh RuntimeProfileRefreshNotifier
var daemonWorkspaceRefresh WorkspaceSetRefreshNotifier
var daemonRuntimeGone RuntimeGoneNotifier
if daemonHub != nil {
daemonProfileRefresh = daemonHub
daemonWorkspaceRefresh = daemonHub
daemonRuntimeGone = daemonHub
}
llmClient := llm.New(llm.Config{
APIKey: cfg.LLMAPIKey,
BaseURL: cfg.LLMBaseURL,
DefaultModel: cfg.LLMDefaultModel,
MaxRetries: cfg.LLMMaxRetries,
DisableThinking: cfg.LLMDisableThinking,
})
// Report the effective retry policy so an operator can confirm from the
// boot log alone what a misbehaving upstream will cost, instead of inferring
// it from an env var whose semantics used to be unguessable (MUL-6364).
// Read off the client, not off cfg, so the line cannot drift from what the
// SDK actually enforces. Counts and an enum only — never the key or the base
// URL, since a self-hosted gateway URL routinely embeds a token. The
// disable_thinking field rides along so the effective request shape is on
// the same line.
llmRetry := llmClient.RetryBudget()
slog.Info("llm retry policy",
"max_retries", llmRetry.MaxRetries,
"source", llmRetry.Source,
"request_timeout", llmRetry.RequestTimeout,
"enabled", llmClient.Enabled(),
"disable_thinking", llmClient.DisableThinking(),
)
taskSvc := service.NewTaskService(queries, txStarter, hub, bus, daemonHub)
taskSvc.Analytics = analyticsClient
taskSvc.SourceContextStorage = store
// Chat follow-up suggestions run through the same internal LLM layer that
// backs auto-titling. A deployment with no MULTICA_LLM_* configuration gets
// a disabled client, which turns the feature off rather than failing.
taskSvc.QuickActions = llmClient
h := &Handler{
Queries: queries,
ReadSelector: dbreader.NewPrimaryOnly(queries),
DB: executor,
TxStarter: txStarter,
Hub: hub,
DaemonHub: daemonHub,
DaemonProfileRefresh: daemonProfileRefresh,
DaemonWorkspaceRefresh: daemonWorkspaceRefresh,
DaemonRuntimeGone: daemonRuntimeGone,
Bus: bus,
TaskService: taskSvc,
PluginService: service.NewPluginService(queries, txStarter),
IssueService: service.NewIssueService(queries, txStarter, bus, analyticsClient, taskSvc),
AutopilotService: service.NewAutopilotService(queries, txStarter, bus, taskSvc),
EmailService: emailService,
UpdateStore: NewInMemoryUpdateStore(),
ModelListStore: NewInMemoryModelListStore(),
ModelCatalogCache: NewInMemoryModelCatalogCache(),
LocalSkillListStore: NewInMemoryLocalSkillListStore(),
LocalSkillImportStore: NewInMemoryLocalSkillImportStore(),
LivenessStore: NewNoopLivenessStore(),
HeartbeatScheduler: NewPassthroughHeartbeatScheduler(queries),
Storage: store,
CFSigner: cfSigner,
Analytics: analyticsClient,
QoderRateLimiter: NewMemoryWebhookRateLimiter(WebhookRateLimit{Limit: 6, Window: time.Minute}),
WebhookRateLimiter: NewMemoryWebhookRateLimiter(DefaultWebhookRateLimit()),
WebhookIPRateLimiter: NewMemoryWebhookIPRateLimiter(DefaultWebhookIPRateLimit()),
WebhookAbsoluteIPRateLimiter: NewMemoryWebhookAbsoluteIPRateLimiter(DefaultWebhookAbsoluteIPRateLimit()),
InvitationRateLimiters: NewMemoryInvitationRateLimiters(DefaultInvitationRateLimits()),
CloudRuntime: cloudruntime.NewClient(cloudruntime.Config{
BaseURL: cfg.CloudURL,
Timeout: cfg.CloudTimeout,
}),
LLM: llmClient,
cfg: cfg,
}
h.WebhookDeliveryWorker = NewWebhookDeliveryWorker(h)
// The default passthrough scheduler reports sweeper-race recoveries so the
// daemon:register refresh fires even without the production batched wiring.
if passthrough, ok := h.HeartbeatScheduler.(*PassthroughHeartbeatScheduler); ok {
passthrough.RecoveryNotifier = h
}
// GitHub API snapshot pipeline for PR cards (MUL-5265). Built
// unconditionally but inert (every trigger no-ops) when the App private key
// is unconfigured, so the feature degrades cleanly. main.go calls
// h.PRRefresh.Start(ctx) to launch its worker pool + TTL sweeper.
ghClient, err := ghsnapshot.NewClientFromEnv()
if err != nil {
// Malformed key is operator-actionable; the pipeline stays disabled.
slog.Warn("github: PR snapshot pipeline disabled (invalid App private key)", "err", err)
}
h.PRRefresh = ghsnapshot.NewManager(ghClient, queries, txStarter, h.broadcastPRSnapshotApplied)
taskSvc.PRSnapshotsEnabled = h.PRRefresh.Enabled()
return h
}
func writeJSON(w http.ResponseWriter, status int, v any) {
// Marshal the body up front so we can advertise an accurate Content-Length
// header. Streaming straight into the ResponseWriter after WriteHeader forces
// net/http into chunked transfer encoding, which omits Content-Length; buffering
// first lets clients (and proxies) see the exact body size.
body, err := json.Marshal(v)
if err != nil {
// Fall back to a minimal, self-describing error payload rather than leaving
// the client with a half-written response.
body = []byte(`{"error":"failed to encode response"}`)
status = http.StatusInternalServerError
}
// Match the trailing newline that json.Encoder.Encode historically appended.
body = append(body, '\n')
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Content-Length", strconv.Itoa(len(body)))
w.WriteHeader(status)
_, _ = w.Write(body)
}
// writeMeasuredJSON behaves like writeJSON but returns the encoded body size so
// callers can record payload bytes in slow-endpoint diagnostics. It measures the
// uncompressed JSON length and is unrelated to transport compression.
func writeMeasuredJSON(w http.ResponseWriter, status int, v any) (int, error) {
body, err := json.Marshal(v)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to encode response")
return 0, err
}
body = append(body, '\n')
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Content-Length", strconv.Itoa(len(body)))
w.WriteHeader(status)
if _, err := w.Write(body); err != nil {
return len(body), err
}
return len(body), nil
}
func writeError(w http.ResponseWriter, status int, msg string) {
writeJSON(w, status, map[string]string{"error": msg})
}
// writeErrorCode is writeError plus a stable machine-readable code, so a UI
// can translate the failure instead of toasting the English sentence at a
// user whose console is in another language. The sentence stays as the
// fallback for anything that has not been given a translation yet.
func writeErrorCode(w http.ResponseWriter, status int, code, msg string) {
writeJSON(w, status, map[string]string{"error": msg, "code": code})
}
// writeFeatureDisabled reports a deliberate deployment or feature gate as a
// non-retryable refusal. A disabled capability is not a transient service
// failure: returning 503 would invite retries and pollute availability alerts.
func writeFeatureDisabled(w http.ResponseWriter, code, msg string) {
writeErrorCode(w, http.StatusForbidden, code, msg)
}
func writeRevisionConflict(w http.ResponseWriter, resourceType string, resourceID pgtype.UUID, expected, actual int64) {
writeJSON(w, http.StatusConflict, map[string]any{
"error": "resource changed since it was loaded",
"code": "revision_conflict",
"resource_type": resourceType,
"resource_id": uuidToString(resourceID),
"expected_revision": expected,
"actual_revision": actual,
})
}
func writeEditConflict(w http.ResponseWriter, resourceType string, resourceID pgtype.UUID) {
writeJSON(w, http.StatusConflict, map[string]any{
"error": "resource field changed since it was loaded",
"code": "revision_conflict",
"resource_type": resourceType,
"resource_id": uuidToString(resourceID),
})
}
// Thin wrappers around util functions.
//
// parseUUID is intentionally the panicking variant: any handler call site
// reachable here is expected to feed a UUID that is either (a) a sqlc round-trip
// of a DB-sourced value, or (b) a raw request input that has already been
// validated upstream. A panic here means an unguarded user-input string slipped
// in — that is a real bug we want surfaced loudly (chi's middleware.Recoverer
// converts it to a 500) instead of silently corrupting data via a zero UUID.
//
// For unvalidated user input at request boundaries, use parseUUIDOrBadRequest
// (writes 400) — never feed raw chi.URLParam / request-body strings into
// parseUUID directly when the call writes to the database.
func parseUUID(s string) pgtype.UUID { return util.MustParseUUID(s) }
func uuidToString(u pgtype.UUID) string { return util.UUIDToString(u) }
func textToPtr(t pgtype.Text) *string { return util.TextToPtr(t) }
func ptrToText(s *string) pgtype.Text { return util.PtrToText(s) }
func strToText(s string) pgtype.Text { return util.StrToText(s) }
func timestampToString(t pgtype.Timestamptz) string { return util.TimestampToString(t) }
func timestampToPtr(t pgtype.Timestamptz) *string { return util.TimestampToPtr(t) }
func timestampToNanoPtr(t pgtype.Timestamptz) *string {
return util.TimestampToNanoPtr(t)
}
func dateToPtr(d pgtype.Date) *string { return util.DateToPtr(d) }
func uuidToPtr(u pgtype.UUID) *string { return util.UUIDToPtr(u) }
// uuidsToStrings maps a UUID array column to string ids, skipping NULL/invalid
// entries. Returns nil (not an empty slice) when there is nothing to emit so
// `omitempty` JSON fields drop out cleanly (MUL-4195).
func uuidsToStrings(us []pgtype.UUID) []string {
if len(us) == 0 {
return nil
}
out := make([]string, 0, len(us))
for _, u := range us {
if u.Valid {
out = append(out, uuidToString(u))
}
}
if len(out) == 0 {
return nil
}
return out
}
// uuidStringsOrEmpty preserves the distinction between a modern, authoritative
// empty UUID-array value (`[]`) and a field omitted by a legacy server. Delivery
// receipts use this so clients never mistake zero delivered comments for an
// unknown receipt and fall back to the enqueue-time plan.
func uuidStringsOrEmpty(us []pgtype.UUID) []string {
out := uuidsToStrings(us)
if out == nil {
return []string{}
}
return out
}
func int8ToPtr(v pgtype.Int8) *int64 { return util.Int8ToPtr(v) }
func int4ToPtr(v pgtype.Int4) *int32 { return util.Int4ToPtr(v) }
func ptrToInt4(v *int32) pgtype.Int4 { return util.PtrToInt4(v) }
// parseUUIDOrBadRequest validates a UUID string sourced from user input
// (URL params, request body, headers). On invalid input it writes a 400
// response and returns ok=false; callers must return immediately.
//
// Use this anywhere a malformed UUID would otherwise reach a write query
// (DELETE / UPDATE) — the silent zero-UUID behavior of the old ParseUUID
// caused real silent-data-loss bugs (#1661).
func parseUUIDOrBadRequest(w http.ResponseWriter, s, fieldName string) (pgtype.UUID, bool) {
u, err := util.ParseUUID(s)
if err != nil {
writeError(w, http.StatusBadRequest, "invalid "+fieldName)
return pgtype.UUID{}, false
}
return u, true
}
func parseUUIDSliceOrBadRequest(w http.ResponseWriter, ids []string, fieldName string) ([]pgtype.UUID, bool) {
uuids := make([]pgtype.UUID, len(ids))
for i, id := range ids {
u, err := util.ParseUUID(id)
if err != nil {
writeError(w, http.StatusBadRequest, "invalid "+fieldName)
return nil, false
}
uuids[i] = u
}
return uuids, true
}
// DeclareChannelFileDelivery records that this deployment can put a file the
// agent produced into a channelType conversation. Call it at boot, from the
// same branch that gives the adapter what it needs to perform the delivery —
// the point is that one condition produces both the capability and the claim
// it is announced under, so a deployment cannot end up promising a hop it does
// not have.
//
// Not safe to call once the server is serving: the map it writes is read
// without a lock by every task claim.
func (h *Handler) DeclareChannelFileDelivery(channelType string) {
if channelType == "" {
return
}
if h.channelFileDelivery == nil {
h.channelFileDelivery = map[string]bool{}
}
h.channelFileDelivery[channelType] = true
}
// channelDeliversFiles answers chat_channel_delivers_files for one claim.
//
// The default is false in every direction that is not an explicit declaration:
// a channel nobody declared, a deployment with no object storage, an adapter
// that was never wired. False is the safe answer because of what the two
// answers cost. False when the file could have travelled loses a delivery the
// agent then describes in words. True when it cannot has the agent write "the
// chart is attached" into a room where nothing is attached, and the reader is
// left hunting for it.
//
// Web chat (empty channel type) is not answered here. It has no adapter and no
// last hop to have — the browser renders the attachment card off the same bind
// — and the prompt handles it in its own branch.
func (h *Handler) channelDeliversFiles(channelType string) bool {
if channelType == "" {
return false
}
return h.channelFileDelivery[channelType]
}
// publish sends a domain event through the event bus.
func (h *Handler) publish(eventType, workspaceID, actorType, actorID string, payload any) {
h.Bus.Publish(events.Event{
Type: eventType,
WorkspaceID: workspaceID,
ActorType: actorType,
ActorID: actorID,
Payload: payload,
})
}
func (h *Handler) notifyDaemonWorkspacesChanged(userIDs ...string) {
if h.DaemonWorkspaceRefresh == nil {
return
}
seen := make(map[string]struct{}, len(userIDs))
for _, userID := range userIDs {
if userID == "" {
continue
}
if _, ok := seen[userID]; ok {
continue
}
seen[userID] = struct{}{}
h.DaemonWorkspaceRefresh.NotifyWorkspacesChanged(userID)
}
}
// NotifyRuntimeGone emits the post-commit runtime invalidation signal. It is
// exported so the runtime GC can use the same publisher as request handlers.
func (h *Handler) NotifyRuntimeGone(runtimeID string) {
if h == nil || h.DaemonRuntimeGone == nil || runtimeID == "" {
return
}
h.DaemonRuntimeGone.NotifyRuntimeGone(runtimeID)
}
// NotifyRuntimeRecovered republishes the workspace-scoped daemon:register
// lifecycle event after a heartbeat restored a runtime from offline. Runtime
// details are intentionally omitted from the payload.
func (h *Handler) NotifyRuntimeRecovered(_ context.Context, workspaceID string) {
if h == nil || workspaceID == "" {
return
}
h.PublishRuntimeRefresh(workspaceID, "system", "", "heartbeat_recovery")
}
// publishTask is publish() plus a TaskID hint so the realtime layer can route
// the event to the per-task scope rather than the whole workspace.
func (h *Handler) publishTask(eventType, workspaceID, actorType, actorID, taskID string, payload any) {
h.Bus.Publish(events.Event{
Type: eventType,
WorkspaceID: workspaceID,
ActorType: actorType,
ActorID: actorID,
TaskID: taskID,
Payload: payload,
})
}
// publishChat is publish() plus the chat session routing hints. actorID must
// be the session's creator: every chat endpoint gates on creator ownership
// before publishing, and the realtime layer delivers the event to that user
// only.
func (h *Handler) publishChat(eventType, workspaceID, actorType, actorID, chatSessionID string, payload any) {
h.Bus.Publish(events.Event{
Type: eventType,
WorkspaceID: workspaceID,
ActorType: actorType,
ActorID: actorID,
ChatSessionID: chatSessionID,
ChatSessionOwnerID: actorID,
Payload: payload,
})
}
func isNotFound(err error) bool {
return errors.Is(err, pgx.ErrNoRows)
}
func isUniqueViolation(err error) bool {
var pgErr *pgconn.PgError
return errors.As(err, &pgErr) && pgErr.Code == "23505"
}
// isCheckViolation reports whether err is a PostgreSQL CHECK constraint
// violation (SQLSTATE 23514). Used to translate column-level CHECK failures
// into a 4xx instead of a generic 500.
func isCheckViolation(err error) bool {
var pgErr *pgconn.PgError
return errors.As(err, &pgErr) && pgErr.Code == "23514"
}
func requestUserID(r *http.Request) string {
return r.Header.Get("X-User-ID")
}
// resolveActor determines whether the request is from an agent or a human member.
//
// First-class signal: X-Actor-Source set to "task_token" means the request
// authenticated via an `mat_` task-scoped token. The auth middleware sets
// that header (and stripped any client-supplied value first), so it is
// authoritative — the bound (agent_id, task_id) cannot be forged or
// stripped by the agent process. This is the path MUL-2600 relies on to
// reject agent-process traffic on owner-only endpoints.
//
// Fallback signal: the request MUST carry both X-Agent-ID and a valid
// X-Task-ID, and the task must belong to the claimed agent. Otherwise we
// fall back to "member".
//
// This fallback is NOT a security boundary and must not be read as one. Both
// ids are observable by any workspace member (GET /api/issues/{id}/task-runs
// returns the pair), so requiring both proves nothing on its own — the older
// comment here claimed it "closes the impersonation path", and it did not.
// What closes it is the Auth / DaemonAuth middleware stripping both headers
// from every client, leaving the mat_ branch as the only writer (MUL-3428).
// The fallback survives that strip only for in-process callers and handler
// unit tests, which set the headers directly.
//
// Returns ("agent", agentID) on success, ("member", userID) otherwise.
func (h *Handler) resolveActor(r *http.Request, userID, workspaceID string) (actorType, actorID string) {
if r.Header.Get("X-Actor-Source") == "task_token" {
// Server-set header — the auth middleware stripped whatever the
// client sent and re-stamped X-Agent-ID from the token row. Trust
// it directly without re-querying.
return "agent", r.Header.Get("X-Agent-ID")
}
agentID := r.Header.Get("X-Agent-ID")
if agentID == "" {
return "member", userID
}
taskID := r.Header.Get("X-Task-ID")
if taskID == "" {
slog.Debug("resolveActor: X-Agent-ID present but X-Task-ID missing, refusing to trust agent identity", "agent_id", agentID)
return "member", userID
}
agentUUID, err := util.ParseUUID(agentID)
if err != nil {
slog.Debug("resolveActor: X-Agent-ID is not a valid UUID, falling back to member", "agent_id", agentID)
return "member", userID
}
// Validate the agent exists in the target workspace.
agent, err := h.Queries.GetAgent(r.Context(), agentUUID)
if err != nil || uuidToString(agent.WorkspaceID) != workspaceID {
slog.Debug("resolveActor: X-Agent-ID rejected, agent not found or workspace mismatch", "agent_id", agentID, "workspace_id", workspaceID)
return "member", userID
}
taskUUID, err := util.ParseUUID(taskID)
if err != nil {
slog.Debug("resolveActor: X-Task-ID is not a valid UUID, falling back to member", "task_id", taskID)
return "member", userID
}
task, err := h.Queries.GetAgentTask(r.Context(), taskUUID)
if err != nil || uuidToString(task.AgentID) != agentID {
slog.Debug("resolveActor: X-Task-ID rejected, task not found or agent mismatch", "agent_id", agentID, "task_id", taskID)
return "member", userID
}
return "agent", agentID
}
func requireUserID(w http.ResponseWriter, r *http.Request) (string, bool) {
userID := requestUserID(r)
if userID == "" {
writeError(w, http.StatusUnauthorized, "user not authenticated")
return "", false
}
return userID, true
}
// resolveWorkspaceID returns the workspace UUID for this request. Delegates
// to middleware.ResolveWorkspaceIDFromRequest so middleware-protected routes
// and middleware-less routes (e.g. /api/upload-file) share identical
// resolution behavior — including slug → UUID translation via the DB.
//
// Returns "" when no workspace identifier was provided or a slug was provided
// but doesn't match any workspace.
func (h *Handler) resolveWorkspaceID(r *http.Request) string {
return middleware.ResolveWorkspaceIDFromRequest(r, h.Queries)
}
// ctxMember returns the workspace member from context (set by workspace middleware).
func ctxMember(ctx context.Context) (db.Member, bool) {
return middleware.MemberFromContext(ctx)
}
// ctxWorkspaceID returns the workspace ID from context (set by workspace middleware).
func ctxWorkspaceID(ctx context.Context) string {
return middleware.WorkspaceIDFromContext(ctx)
}
// workspaceIDFromURL returns the workspace ID from context (preferred) or chi URL param (fallback).
func workspaceIDFromURL(r *http.Request, param string) string {
if id := middleware.WorkspaceIDFromContext(r.Context()); id != "" {
return id
}
return chi.URLParam(r, param)
}
// workspaceMember returns the member from middleware context, or falls back to a DB
// lookup when the handler is called directly (e.g. in tests).
func (h *Handler) workspaceMember(w http.ResponseWriter, r *http.Request, workspaceID string) (db.Member, bool) {
if m, ok := ctxMember(r.Context()); ok {
return m, true
}
return h.requireWorkspaceMember(w, r, workspaceID, "workspace not found")
}
func roleAllowed(role string, roles ...string) bool {
for _, candidate := range roles {
if role == candidate {
return true
}
}
return false
}
func countOwners(members []db.Member) int {
owners := 0
for _, member := range members {
if member.Role == "owner" {
owners++
}
}
return owners
}
func (h *Handler) getWorkspaceMember(ctx context.Context, userID, workspaceID string) (db.Member, error) {
userUUID, err := util.ParseUUID(userID)
if err != nil {
return db.Member{}, err
}
wsUUID, err := util.ParseUUID(workspaceID)
if err != nil {
return db.Member{}, err
}
return h.Queries.GetMemberByUserAndWorkspace(ctx, db.GetMemberByUserAndWorkspaceParams{
UserID: userUUID,
WorkspaceID: wsUUID,
})
}
func (h *Handler) requireWorkspaceMember(w http.ResponseWriter, r *http.Request, workspaceID, notFoundMsg string) (db.Member, bool) {
if workspaceID == "" {
writeError(w, http.StatusBadRequest, "workspace_id is required")
return db.Member{}, false
}
userID, ok := requireUserID(w, r)
if !ok {
return db.Member{}, false
}
member, err := h.getWorkspaceMember(r.Context(), userID, workspaceID)
if err != nil {
writeError(w, http.StatusNotFound, notFoundMsg)
return db.Member{}, false
}
return member, true
}
func (h *Handler) requireWorkspaceRole(w http.ResponseWriter, r *http.Request, workspaceID, notFoundMsg string, roles ...string) (db.Member, bool) {
member, ok := h.requireWorkspaceMember(w, r, workspaceID, notFoundMsg)
if !ok {
return db.Member{}, false
}
if !roleAllowed(member.Role, roles...) {
writeError(w, http.StatusForbidden, "insufficient permissions")
return db.Member{}, false
}
return member, true
}
// isWorkspaceEntity checks whether a user_id belongs to the given workspace,
// as either a member or an agent depending on userType.
func (h *Handler) isWorkspaceEntity(ctx context.Context, userType, userID, workspaceID string) bool {
switch userType {
case "member":
_, err := h.getWorkspaceMember(ctx, userID, workspaceID)
return err == nil
case "agent":
userUUID, err := util.ParseUUID(userID)
if err != nil {
return false
}
wsUUID, err := util.ParseUUID(workspaceID)
if err != nil {
return false
}
_, err = h.Queries.GetAgentInWorkspace(ctx, db.GetAgentInWorkspaceParams{
ID: userUUID,
WorkspaceID: wsUUID,
})
return err == nil
default:
return false
}
}
func (h *Handler) loadIssueForUser(w http.ResponseWriter, r *http.Request, issueID string) (db.Issue, bool) {
if _, ok := requireUserID(w, r); !ok {
return db.Issue{}, false
}
workspaceID := h.resolveWorkspaceID(r)
if workspaceID == "" {
writeError(w, http.StatusBadRequest, "workspace_id is required")
return db.Issue{}, false
}
// Try identifier format first (e.g., "JIA-42"). resolveIssueByIdentifier
// silently returns false for non-identifier strings, falling through to
// the UUID path below.
if issue, ok := h.resolveIssueByIdentifier(r.Context(), issueID, workspaceID); ok {
return issue, true
}
issueUUID, err := util.ParseUUID(issueID)
if err != nil {
// Not a valid UUID and didn't match identifier format → 404 (consistent
// with previous silent-zero behavior, which would also have produced 404).
writeError(w, http.StatusNotFound, "issue not found")
return db.Issue{}, false
}
wsUUID, err := util.ParseUUID(workspaceID)
if err != nil {
writeError(w, http.StatusBadRequest, "invalid workspace_id")
return db.Issue{}, false
}
issue, err := h.Queries.GetIssueInWorkspace(r.Context(), db.GetIssueInWorkspaceParams{
ID: issueUUID,
WorkspaceID: wsUUID,
})
if err != nil {
writeError(w, http.StatusNotFound, "issue not found")
return db.Issue{}, false
}
return issue, true
}
// resolveIssueByIdentifier tries to look up an issue by "PREFIX-NUMBER" format.
//
// The prefix must match the workspace's own issue prefix, the same rule
// `lookupIssueByIdentifier` applies to VCS webhooks. Without it every prefix
// resolved to the same issue number, so `FOO-134` and `TRS-134` were
// interchangeable — which makes the identifier URL `/{ws}/issues/{key}`
// unusable as a canonical link.
func (h *Handler) resolveIssueByIdentifier(ctx context.Context, id, workspaceID string) (db.Issue, bool) {
parts := splitIdentifier(id)
if parts == nil {
return db.Issue{}, false
}
if workspaceID == "" {
return db.Issue{}, false
}
wsUUID, err := util.ParseUUID(workspaceID)
if err != nil {
return db.Issue{}, false
}
// Case-insensitive: a hand-typed `trs-134` should open `TRS-134`.
prefix := h.getIssuePrefix(ctx, wsUUID)
if prefix == "" || !strings.EqualFold(parts.prefix, prefix) {
return db.Issue{}, false
}
issue, err := h.Queries.GetIssueByNumber(ctx, db.GetIssueByNumberParams{
WorkspaceID: wsUUID,
Number: parts.number,
})
if err != nil {
return db.Issue{}, false
}
return issue, true
}
type identifierParts struct {
prefix string
number int32
}
func splitIdentifier(id string) *identifierParts {
idx := -1
for i := len(id) - 1; i >= 0; i-- {
if id[i] == '-' {
idx = i
break
}
}
if idx <= 0 || idx >= len(id)-1 {
return nil
}
numStr := id[idx+1:]
num := 0
for _, c := range numStr {
if c < '0' || c > '9' {
return nil
}
num = num*10 + int(c-'0')
// Guard the int32 conversion below: a UUID whose last group happens to
// be all digits ("…-421234567890") reaches here and would otherwise be
// truncated into a plausible-looking issue number.
if num > math.MaxInt32 {
return nil
}
}
if num <= 0 {
return nil
}
return &identifierParts{prefix: id[:idx], number: int32(num)}
}
// issuePrefixForWorkspace resolves a workspace row's effective issue prefix:
// the configured value, or one generated from the workspace name when the
// stored prefix is empty (e.g. workspaces created before the prefix was
// introduced). Split out from getIssuePrefix so callers that already hold the
// row — such as the GitHub close-intent scan, which must not re-read it — can
// reuse the rule.
//
// The empty-prefix fallback stays on the FROZEN name-based derivation, not the
// slug-based one new workspaces get (MUL-6050): identifiers are computed at
// read time, so switching this path would rewrite the identifier of every
// issue in those legacy workspaces. New workspaces always persist a prefix at
// creation, so they never reach this branch.
func issuePrefixForWorkspace(ws db.Workspace) string {
if ws.IssuePrefix != "" {
return ws.IssuePrefix
}
return legacyIssuePrefixFromName(ws.Name)
}
// getIssuePrefix fetches the effective issue_prefix for a workspace, and
// returns "" when the workspace row cannot be loaded.
func (h *Handler) getIssuePrefix(ctx context.Context, workspaceID pgtype.UUID) string {
ws, err := h.Queries.GetWorkspace(ctx, workspaceID)
if err != nil {
return ""
}
return issuePrefixForWorkspace(ws)
}
func (h *Handler) loadAgentForUser(w http.ResponseWriter, r *http.Request, agentID string) (db.Agent, bool) {
if _, ok := requireUserID(w, r); !ok {
return db.Agent{}, false
}
workspaceID := h.resolveWorkspaceID(r)
if workspaceID == "" {
writeError(w, http.StatusBadRequest, "workspace_id is required")
return db.Agent{}, false
}
agentUUID, ok := parseUUIDOrBadRequest(w, agentID, "agent id")
if !ok {
return db.Agent{}, false
}
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
if !ok {
return db.Agent{}, false
}
agent, err := h.Queries.GetAgentInWorkspace(r.Context(), db.GetAgentInWorkspaceParams{
ID: agentUUID,
WorkspaceID: wsUUID,
})
if err != nil {
writeError(w, http.StatusNotFound, "agent not found")
return db.Agent{}, false
}
if agent.Kind != "user" {
writeError(w, http.StatusNotFound, "agent not found")
return db.Agent{}, false
}
return agent, true
}
func (h *Handler) loadInboxItemForUser(w http.ResponseWriter, r *http.Request, itemID string) (db.InboxItem, bool) {
userID, ok := requireUserID(w, r)
if !ok {
return db.InboxItem{}, false
}
workspaceID := h.resolveWorkspaceID(r)
if workspaceID == "" {
writeError(w, http.StatusBadRequest, "workspace_id is required")
return db.InboxItem{}, false
}
itemUUID, ok := parseUUIDOrBadRequest(w, itemID, "inbox item id")
if !ok {
return db.InboxItem{}, false
}
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
if !ok {
return db.InboxItem{}, false
}
item, err := h.Queries.GetInboxItemInWorkspace(r.Context(), db.GetInboxItemInWorkspaceParams{
ID: itemUUID,
WorkspaceID: wsUUID,
})
if err != nil {
writeError(w, http.StatusNotFound, "inbox item not found")
return db.InboxItem{}, false
}
if item.RecipientType != "member" || uuidToString(item.RecipientID) != userID {
writeError(w, http.StatusNotFound, "inbox item not found")
return db.InboxItem{}, false
}
return item, true
}