Files
silo-server/internal/api/handlers/user_collection_imports.go
203a18ae83 feat(observability): OpenTelemetry logs+traces with secret redaction and slog standardization (#290)
* feat(observability): OpenTelemetry logs+traces with secret redaction

Part of #265. Adds opt-in OpenTelemetry (logs + traces) alongside the existing
stderr + opslog pipeline, plus secret redaction on all sinks. Default-off: with
no OTEL_* / SILO_OTEL_ENABLED config, behavior is unchanged.

Bootstrap (internal/telemetry):
- Setup() builds one shared resource, a TracerProvider (parent-based trace-id
  ratio sampler), a LoggerProvider, and the W3C TraceContext+Baggage propagator
  from env. It installs NO MeterProvider — metrics stay on Prometheus, and the
  built-in no-op global MeterProvider keeps the trace instrumentation libs from
  double-emitting. Shutdown is deferred with a flush timeout.
- Logs are bridged via otelslog fan-out (slog.MultiHandler), level-gated by the
  shared LevelVar and best-effort so a failing collector can't break the console
  or DB branches. stderr + opslog stay untouched.

Secret redaction (internal/logredact):
- A slog.Handler masks secret-keyed attributes (password, token, api_key,
  authorization, cookie, ...) — including .With-bound attrs, nested groups,
  secret-keyed group subtrees, and values behind a LogValuer — on the console
  and OTLP sinks, with a no-op fast path when a record has no secret keys.
  opslog.shouldRedact delegates to logredact.SecretKey so all sinks share one
  marker list.

Rotation is infra-managed (no custom file sink): container runtime for stderr,
collector/backend for OTLP, opslog partition-pruning for the DB. Documented in
docs/architecture/observability.md.

Verification: go build ./..., go vet, gofmt -l — clean; go test
./internal/telemetry/ ./internal/logredact/ -race pass.

AI-use disclosure: implemented with AI assistance (Claude Code), including
adversarial reviews that hardened the bootstrap and fixed two redaction leak
paths; reviewed by the author.

* refactor(observability): slog context+component sweep, sloglint gate (phase 3)

Part of #265. Builds on the OTel bootstrap + redaction commit.

Standardizes every log call site onto the context-carrying slog variants so
records correlate with the active OpenTelemetry trace, and locks the standard
in with a machine gate so future code (human- or AI-authored) can't drift back.

- Call-site sweep: converted the remaining slog.<Level>(...) calls to the
  slog.<Level>Context(ctx, ...) form wherever a context.Context is in scope
  (background/init calls with no ctx are left as-is), across 183 files. Applied
  via a type-aware AST codemod. Log levels and message strings are preserved
  verbatim; a component attr (canonical per-package name) is added to direct
  package-level slog calls. Bound-logger calls keep their existing .With
  bindings. The main.go and telemetry package conversions rode with their file
  in the previous commit to keep each file within a single commit.
- Enforcement (.golangci.yml): enable sloglint with context=scope, static-msg,
  key-naming-case=snake, no-mixed-args. After the sweep all four report zero
  violations repo-wide (tests included), so make lint / CI now blocks any
  regression to the non-context form. The gate ships with the sweep because it
  cannot be green until the legacy sites are converted.

Metrics remain on Prometheus; no behavior change to /metrics or Grafana.

Verification: go build ./..., go vet ./..., gofmt -l — clean; sloglint (all 4
rules) 0 violations repo-wide; log levels verified unchanged.

AI-use disclosure: implemented with AI assistance (Claude Code), including the
codemod; reviewed by the author.

* fix(observability): honor per-signal OTLP protocol and secret WithGroup names

Two Codex review findings on PR #290:

- telemetry: OTEL_EXPORTER_OTLP_{TRACES,LOGS}_PROTOCOL now override the
  generic OTEL_EXPORTER_OTLP_PROTOCOL per signal, so mixed collector
  setups (e.g. HTTP logs + gRPC traces) build the right exporter.
- logredact: entering a group whose name is secret-bearing (e.g.
  WithGroup("authorization")) now masks every leaf in that subtree,
  matching how slog.Group("authorization", ...) is masked as a whole.

* fix(observability): address review feedback on telemetry bootstrap

- Telemetry setup failure no longer kills boot: Setup returns usable
  no-op providers alongside the error and main logs and continues with
  telemetry disabled, honoring the best-effort contract.
- Honor OTEL_TRACES_SAMPLER (always_on/off, traceidratio, parentbased_*
  variants); unsupported values fall back to parentbased_traceidratio.
- Attach node identity as semconv service.instance.id instead of the
  non-semconv node.name.
- Rename opslog retention-scope log attrs to target_component/target_level
  so they no longer collide with the canonical component routing key, and
  tag those lines with component=opslog.
- Fix stale levelGated comment casing; use WarnContext in the telemetry
  shutdown defer; document the LogValuer double-resolve on the redaction
  slow path.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Quick <31828688+Quick104@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-09 08:53:52 -04:00

451 lines
14 KiB
Go

package handlers
import (
"context"
"encoding/json"
"errors"
"fmt"
"io/fs"
"log/slog"
"net/http"
"strings"
"time"
"github.com/go-chi/chi/v5"
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
"github.com/Silo-Server/silo-server/internal/catalog"
"github.com/Silo-Server/silo-server/internal/collections/templates"
"github.com/Silo-Server/silo-server/internal/collectionutil"
"github.com/Silo-Server/silo-server/internal/mdblist"
"github.com/Silo-Server/silo-server/internal/s3client"
"github.com/Silo-Server/silo-server/internal/usercollections"
"github.com/Silo-Server/silo-server/internal/userstore"
)
// UserCollectionImportHandler exposes the user-side template gallery and
// import + sync endpoints. Authorization rules (only the creator can sync) are
// enforced here; the underlying sync.Service is intentionally unauthenticated.
type UserCollectionImportHandler struct {
storeProvider userstore.UserStoreProvider
sync *usercollections.Service
scheduler *usercollections.Scheduler
registry *templates.Registry
mdblist *mdblist.Client
s3GP *s3client.Client
frontendFS fs.FS
presignTTL time.Duration
}
func NewUserCollectionImportHandler(
provider userstore.UserStoreProvider,
sync *usercollections.Service,
scheduler *usercollections.Scheduler,
registry *templates.Registry,
mdblistClient *mdblist.Client,
s3GP *s3client.Client,
frontendFS fs.FS,
presignTTL time.Duration,
) *UserCollectionImportHandler {
if registry == nil {
registry = templates.Default
}
return &UserCollectionImportHandler{
storeProvider: provider,
sync: sync,
scheduler: scheduler,
registry: registry,
mdblist: mdblistClient,
s3GP: s3GP,
frontendFS: frontendFS,
presignTTL: presignTTL,
}
}
func (h *UserCollectionImportHandler) HandleListTemplates(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, h.registry.Catalog())
}
type userImportSharedFields struct {
Title string `json:"title"`
Description string `json:"description"`
Limit *int `json:"limit,omitempty"`
SyncSchedule string `json:"sync_schedule"`
IsShared bool `json:"is_shared"`
PosterURL string `json:"poster_url"`
LibraryIDs []int `json:"library_ids,omitempty"`
DisplayQueryDefinition json.RawMessage `json:"display_query_definition,omitempty"`
}
type userImportMDBListRequest struct {
userImportSharedFields
URL string `json:"url"`
}
type userImportTMDBRequest struct {
userImportSharedFields
Preset string `json:"preset"`
MediaType string `json:"media_type"`
TimeWindow string `json:"time_window"`
}
type userImportTraktRequest struct {
userImportSharedFields
Preset string `json:"preset"`
MediaType string `json:"media_type"`
}
type userImportResponse struct {
Collection collectionResponse `json:"collection"`
Sync *usercollections.SyncResult `json:"sync,omitempty"`
}
func (h *UserCollectionImportHandler) HandleImportMDBList(w http.ResponseWriter, r *http.Request) {
var req userImportMDBListRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
return
}
if strings.TrimSpace(req.Title) == "" || strings.TrimSpace(req.URL) == "" {
writeError(w, http.StatusBadRequest, "bad_request", "title and url are required")
return
}
if !validateOptionalLimit(req.Limit, w) {
return
}
cfg := usercollections.SourceConfig{
Mode: usercollections.SourceModeMDBList,
URL: usercollections.NormalizeMDBListURL(req.URL),
Limit: req.Limit,
LibraryIDs: req.LibraryIDs,
}
h.createImportedCollection(w, r, "mdblist", cfg, req.userImportSharedFields)
}
func (h *UserCollectionImportHandler) HandleImportTMDB(w http.ResponseWriter, r *http.Request) {
var req userImportTMDBRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
return
}
if strings.TrimSpace(req.Title) == "" {
writeError(w, http.StatusBadRequest, "bad_request", "title is required")
return
}
preset, mediaType, timeWindow, err := normalizeTMDBPresetRequest(req.Preset, req.MediaType, req.TimeWindow)
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
if !validateOptionalLimit(req.Limit, w) {
return
}
cfg := usercollections.SourceConfig{
Mode: usercollections.SourceModeTMDBPreset,
Preset: preset,
MediaType: mediaType,
TimeWindow: timeWindow,
Limit: req.Limit,
LibraryIDs: req.LibraryIDs,
}
h.createImportedCollection(w, r, "tmdb", cfg, req.userImportSharedFields)
}
func (h *UserCollectionImportHandler) HandleImportTrakt(w http.ResponseWriter, r *http.Request) {
var req userImportTraktRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
return
}
if strings.TrimSpace(req.Title) == "" {
writeError(w, http.StatusBadRequest, "bad_request", "title is required")
return
}
profileID := apimw.GetProfileID(r.Context())
preset, mediaType, normalizedProfileID, err := normalizeTraktPresetRequest(req.Preset, req.MediaType, profileID)
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
if !validateOptionalLimit(req.Limit, w) {
return
}
cfg := usercollections.SourceConfig{
Mode: usercollections.SourceModeTraktPreset,
Provider: "trakt",
Preset: preset,
MediaType: mediaType,
ProfileID: normalizedProfileID,
Limit: req.Limit,
LibraryIDs: req.LibraryIDs,
}
h.createImportedCollection(w, r, "trakt", cfg, req.userImportSharedFields)
}
func (h *UserCollectionImportHandler) createImportedCollection(
w http.ResponseWriter,
r *http.Request,
collectionType string,
cfg usercollections.SourceConfig,
shared userImportSharedFields,
) {
userID := apimw.GetUserID(r.Context())
profileID := apimw.GetProfileID(r.Context())
store, err := h.storeProvider.ForUser(r.Context(), userID)
if err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to access user store")
return
}
schedule, err := usercollections.ResolveSyncSchedule(shared.SyncSchedule)
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
if !validateOptionalLibraryIDs(cfg.LibraryIDs, w) {
return
}
sourceConfigJSON, err := usercollections.MarshalSourceConfig(cfg)
if err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to encode source config")
return
}
displayQueryDefinition, err := catalog.NormalizeDisplayQueryFragment(shared.DisplayQueryDefinition)
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", err.Error())
return
}
collection, err := store.CreateCollection(r.Context(), userstore.CreateCollectionInput{
CreatorProfileID: profileID,
Name: strings.TrimSpace(shared.Title),
Description: strings.TrimSpace(shared.Description),
CollectionType: collectionType,
IsShared: shared.IsShared,
QueryDefinition: "{}",
SortConfig: "{}",
SourceURL: cfg.DisplayURL(),
SourceConfig: sourceConfigJSON,
SyncSchedule: schedule,
NextSyncAt: usercollections.InitialNextSyncAt(schedule),
DisplayQueryDefinition: displayQueryDefinition,
PosterURL: strings.TrimSpace(shared.PosterURL),
})
if err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to create collection")
return
}
if err := h.storeBundledTemplatePoster(r, store, collection, strings.TrimSpace(shared.PosterURL)); err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to process template poster")
return
}
syncResult, updated, syncErr := h.sync.RunSync(r.Context(), store, collection)
if syncErr != nil {
// Persist failure state inline so the UI shows the error and the user
// can retry; the row is intentionally kept around for that retry path.
_ = store.UpdateCollectionSyncState(r.Context(), userstore.UpdateCollectionSyncStateInput{
ID: collection.ID,
Status: "failed",
Message: syncErr.Error(),
LastSyncAt: time.Now().UTC(),
NextSyncAt: usercollections.InitialNextSyncAt(schedule),
})
updated = collection
updated.LastSyncStatus = "failed"
updated.LastSyncMessage = syncErr.Error()
}
writeJSON(w, http.StatusCreated, userImportResponse{
Collection: h.toCollectionResponse(r, *updated),
Sync: syncResult,
})
}
func (h *UserCollectionImportHandler) storeBundledTemplatePoster(
r *http.Request,
store userstore.UserStore,
collection *userstore.Collection,
posterPath string,
) error {
if collection == nil {
return nil
}
storedPath, thumbhash, stored, err := storeBundledCollectionPosterIfS3Configured(
r.Context(),
h.s3GP,
h.frontendFS,
collection.ID,
userCollectionImagePrefix,
posterPath,
)
if err != nil || !stored {
if err != nil {
slog.WarnContext(r.Context(), "failed to store bundled user collection poster", "component", "api",
"collection_id", collection.ID,
"poster_path", posterPath,
"error", err,
)
}
return nil
}
if err := store.UpdateCollection(r.Context(), userstore.UpdateCollectionInput{
ID: collection.ID,
RequestProfileID: collection.CreatorProfileID,
PosterURL: &storedPath,
PosterThumbhash: &thumbhash,
}); err != nil {
slog.WarnContext(r.Context(), "failed to persist bundled user collection poster", "component", "api",
"collection_id", collection.ID,
"poster_path", posterPath,
"stored_path", storedPath,
"error", err,
)
return nil
}
collection.PosterURL = storedPath
collection.PosterThumbhash = thumbhash
return nil
}
func (h *UserCollectionImportHandler) toCollectionResponse(r *http.Request, c userstore.Collection) collectionResponse {
resp := toCollectionResponse(c)
resp.PosterURL = h.presignCollectionPoster(r.Context(), c.PosterURL)
return resp
}
func (h *UserCollectionImportHandler) presignCollectionPoster(ctx context.Context, path string) string {
if path == "" {
return ""
}
if strings.HasPrefix(path, "http://") || strings.HasPrefix(path, "https://") {
return path
}
if strings.HasPrefix(path, "/") {
return path
}
if h.s3GP == nil {
return ""
}
ttl := h.presignTTL
if ttl <= 0 {
ttl = 4 * time.Hour
}
url, err := h.s3GP.PresignGetURL(ctx, h.s3GP.Bucket(), cardThumbnailPath(path), ttl)
if err != nil {
return ""
}
return url
}
func (h *UserCollectionImportHandler) HandleSync(w http.ResponseWriter, r *http.Request) {
collectionID := chi.URLParam(r, "id")
if collectionID == "" {
writeError(w, http.StatusBadRequest, "bad_request", "Collection ID is required")
return
}
userID := apimw.GetUserID(r.Context())
profileID := apimw.GetProfileID(r.Context())
store, err := h.storeProvider.ForUser(r.Context(), userID)
if err != nil {
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to access user store")
return
}
collection, err := store.GetCollection(r.Context(), collectionID)
if err != nil {
writeError(w, http.StatusNotFound, "not_found", "Collection not found")
return
}
if collection.CreatorProfileID != profileID {
writeError(w, http.StatusForbidden, "forbidden", "Only the creator can sync this collection")
return
}
if h.scheduler != nil && h.scheduler.IsInFlight(collectionID) {
writeError(w, http.StatusConflict, "sync_in_flight", "A sync is already running for this collection")
return
}
result, _, err := h.sync.RunSync(r.Context(), store, collection)
if err != nil {
if errors.Is(err, usercollections.ErrSyncUnsupported) {
writeError(w, http.StatusBadRequest, "bad_request", "This collection does not support sync")
return
}
writeError(w, http.StatusInternalServerError, "internal_error", fmt.Sprintf("Sync failed: %v", err))
return
}
writeJSON(w, http.StatusOK, result)
}
type mdblistDiscoveryResponse struct {
Configured bool `json:"configured"`
Lists []mdblist.ListSummary `json:"lists"`
}
// mdblistConfigured returns true when the discovery client is usable. When
// false it has already written a "not configured" 200 response so callers
// can simply early-return.
func (h *UserCollectionImportHandler) mdblistConfigured(w http.ResponseWriter) bool {
if h.mdblist != nil && h.mdblist.Configured() {
return true
}
writeJSON(w, http.StatusOK, mdblistDiscoveryResponse{Configured: false, Lists: []mdblist.ListSummary{}})
return false
}
func (h *UserCollectionImportHandler) HandleSearchMDBList(w http.ResponseWriter, r *http.Request) {
if !h.mdblistConfigured(w) {
return
}
query := strings.TrimSpace(r.URL.Query().Get("q"))
if query == "" {
writeError(w, http.StatusBadRequest, "bad_request", "q is required")
return
}
lists, err := h.mdblist.Search(r.Context(), query)
if err != nil {
writeError(w, http.StatusBadGateway, "upstream_error", fmt.Sprintf("MDBList search failed: %v", err))
return
}
writeJSON(w, http.StatusOK, mdblistDiscoveryResponse{Configured: true, Lists: lists})
}
func (h *UserCollectionImportHandler) HandleTopMDBList(w http.ResponseWriter, r *http.Request) {
if !h.mdblistConfigured(w) {
return
}
lists, err := h.mdblist.Top(r.Context())
if err != nil {
writeError(w, http.StatusBadGateway, "upstream_error", fmt.Sprintf("MDBList top failed: %v", err))
return
}
writeJSON(w, http.StatusOK, mdblistDiscoveryResponse{Configured: true, Lists: lists})
}
func validateOptionalLimit(limit *int, w http.ResponseWriter) bool {
if limit == nil {
return true
}
if *limit <= 0 || *limit > collectionutil.MaxExplicitItemLimit {
writeError(w, http.StatusBadRequest, "bad_request",
fmt.Sprintf("limit must be between 1 and %d", collectionutil.MaxExplicitItemLimit))
return false
}
return true
}
func validateOptionalLibraryIDs(libraryIDs []int, w http.ResponseWriter) bool {
for _, id := range libraryIDs {
if id <= 0 {
writeError(w, http.StatusBadRequest, "bad_request", "library_ids must contain positive IDs")
return false
}
}
return true
}