mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-08 01:16:45 +08:00
1273 lines
53 KiB
Go
1273 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"
|
|
"github.com/multica-ai/multica/server/internal/marketcatalog"
|
|
"github.com/multica-ai/multica/server/internal/marketinstall"
|
|
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
|
|
// 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
|
|
MarketCatalogURL string
|
|
MarketCatalogToken string
|
|
// PlatformAdminEmails bootstrap the platform admin console on first use.
|
|
PlatformAdminEmails []string
|
|
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
|
|
}
|
|
|
|
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
|
|
WebhookRateLimiter WebhookRateLimiter
|
|
WebhookIPRateLimiter WebhookRateLimiter
|
|
WebhookAbsoluteIPRateLimiter WebhookRateLimiter
|
|
InvitationRateLimiters InvitationRateLimiters
|
|
WebhookDeliveryWorker *WebhookDeliveryWorker
|
|
CloudRuntime cloudRuntimeProxy
|
|
MarketCatalog *marketcatalog.Client
|
|
MarketInstall *marketinstall.Ledger
|
|
// 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,
|
|
WebhookRateLimiter: NewMemoryWebhookRateLimiter(DefaultWebhookRateLimit()),
|
|
WebhookIPRateLimiter: NewMemoryWebhookIPRateLimiter(DefaultWebhookIPRateLimit()),
|
|
WebhookAbsoluteIPRateLimiter: NewMemoryWebhookAbsoluteIPRateLimiter(DefaultWebhookAbsoluteIPRateLimit()),
|
|
InvitationRateLimiters: NewMemoryInvitationRateLimiters(DefaultInvitationRateLimits()),
|
|
CloudRuntime: cloudruntime.NewClient(cloudruntime.Config{
|
|
BaseURL: cfg.CloudURL,
|
|
Timeout: cfg.CloudTimeout,
|
|
}),
|
|
MarketCatalog: marketcatalog.New(marketcatalog.Config{
|
|
BaseURL: cfg.MarketCatalogURL,
|
|
Token: cfg.MarketCatalogToken,
|
|
}),
|
|
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()
|
|
h.WireMarketInstall()
|
|
|
|
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 a ChatSessionID hint so the realtime layer
|
|
// can route the event to the per-chat-session scope.
|
|
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,
|
|
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
|
|
}
|