diff --git a/internal/api/handlers/admin.go b/internal/api/handlers/admin.go index bf453046..5a884aeb 100644 --- a/internal/api/handlers/admin.go +++ b/internal/api/handlers/admin.go @@ -20,8 +20,8 @@ import ( "golang.org/x/sync/errgroup" "github.com/Silo-Server/silo-server/internal/access" - "github.com/Silo-Server/silo-server/internal/ai/llm" "github.com/Silo-Server/silo-server/internal/adminjob" + "github.com/Silo-Server/silo-server/internal/ai/llm" apimw "github.com/Silo-Server/silo-server/internal/api/middleware" "github.com/Silo-Server/silo-server/internal/auth" "github.com/Silo-Server/silo-server/internal/cache" @@ -30,6 +30,7 @@ import ( "github.com/Silo-Server/silo-server/internal/markers" "github.com/Silo-Server/silo-server/internal/models" "github.com/Silo-Server/silo-server/internal/notifications" + subtitleai "github.com/Silo-Server/silo-server/internal/subtitles/ai" "github.com/Silo-Server/silo-server/internal/userstore" ) @@ -2117,6 +2118,18 @@ func (h *AdminHandler) HandleUpdateSetting(w http.ResponseWriter, r *http.Reques "metadata_ai.on_view must be off, button, or auto") return } + case "subtitle_ai.transcribe_quota_jobs": + if n, err := strconv.Atoi(req.Value); err != nil || n < 0 { + writeError(w, http.StatusBadRequest, "bad_request", + "subtitle_ai.transcribe_quota_jobs must be an integer >= 0 (0 = unlimited)") + return + } + case "subtitle_ai.transcribe_quota_period": + if !subtitleai.ValidQuotaPeriod(req.Value) { + writeError(w, http.StatusBadRequest, "bad_request", + "subtitle_ai.transcribe_quota_period must be day, week, or month") + return + } } if err := h.SettingsRepo.Set(r.Context(), key, req.Value); err != nil { diff --git a/internal/api/handlers/profiles.go b/internal/api/handlers/profiles.go index 07caacd1..9c2a729d 100644 --- a/internal/api/handlers/profiles.go +++ b/internal/api/handlers/profiles.go @@ -124,11 +124,6 @@ type verifyPINResponse struct { ExpiresAt string `json:"expires_at,omitempty"` } -func isAdminProfileRequest(ctx context.Context) bool { - claims := apimw.GetClaims(ctx) - return claims != nil && claims.Role == "admin" -} - // canManageHouseholdProfiles reports whether the caller may create/update/delete // profiles belonging to their user. Server admins always can. Otherwise the // caller's active profile must be the primary profile for the household. @@ -138,7 +133,7 @@ func isAdminProfileRequest(ctx context.Context) bool { // bypass the profile lock by sending only X-Profile-Id. func (h *ProfileHandler) canManageHouseholdProfiles(r *http.Request, store userstore.UserStore) (bool, error) { ctx := r.Context() - if isAdminProfileRequest(ctx) { + if apimw.IsAdmin(ctx) { return true, nil } activeProfileID := apimw.GetProfileID(ctx) @@ -316,7 +311,7 @@ func (h *ProfileHandler) HandleCreateProfile(w http.ResponseWriter, r *http.Requ // Access-policy fields only make sense when set by a manager on a managed // profile. On bootstrap the caller is becoming primary themselves, so non- // admin bootstrap creations must leave those fields at their defaults. - if isBootstrap && !isAdminProfileRequest(r.Context()) && + if isBootstrap && !apimw.IsAdmin(r.Context()) && (req.IsChild || req.MaxContentRating != "" || req.LibraryRestrictionsEnabled || len(req.AllowedLibraryIDs) > 0 || req.MaxPlaybackQuality != "") { diff --git a/internal/api/handlers/subtitle_ai.go b/internal/api/handlers/subtitle_ai.go index 90a8e021..67dd1a1c 100644 --- a/internal/api/handlers/subtitle_ai.go +++ b/internal/api/handlers/subtitle_ai.go @@ -3,6 +3,7 @@ package handlers import ( "encoding/json" "errors" + "fmt" "log/slog" "net/http" "strconv" @@ -12,6 +13,7 @@ import ( 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/subtitles/ai" + "github.com/Silo-Server/silo-server/internal/userstore" ) // SubtitleAIHandler exposes on-demand AI subtitle translation backed by the @@ -21,6 +23,9 @@ import ( type SubtitleAIHandler struct { service *ai.Service FileAuthorizer *MediaFileAuthorizer + // StoreProvider resolves household profiles for the transcription quota + // exemption check; when nil the whole admin account is exempt. + StoreProvider userstore.UserStoreProvider } // NewSubtitleAIHandler creates a handler backed by the given service. @@ -117,6 +122,7 @@ func (h *SubtitleAIHandler) HandleTranslate(w http.ResponseWriter, r *http.Reque SourceLanguage: req.SourceLanguage, TargetLanguage: req.TargetLanguage, RequestedBy: requestedBy, + QuotaExempt: h.quotaExempt(r), SessionID: req.SessionID, StartPosition: req.StartPosition, }) @@ -125,6 +131,8 @@ func (h *SubtitleAIHandler) HandleTranslate(w http.ResponseWriter, r *http.Reque case errors.Is(err, ai.ErrEngineNotConfigured): writeError(w, http.StatusServiceUnavailable, "not_configured", "AI subtitle translation is not configured on this server") + case errors.Is(err, ai.ErrQuotaExceeded): + writeError(w, http.StatusTooManyRequests, "quota_exceeded", quotaExceededMessage(err)) case errors.Is(err, ai.ErrInvalidRequest): writeError(w, http.StatusBadRequest, "bad_request", err.Error()) default: @@ -138,6 +146,60 @@ func (h *SubtitleAIHandler) HandleTranslate(w http.ResponseWriter, r *http.Reque writeJSON(w, http.StatusAccepted, map[string]any{"job": job}) } +// quotaExempt reports whether the caller is exempt from the transcription +// quota: an admin account acting as itself (no profile context) or through +// its primary (household parent) profile. Other profiles on an admin account +// — e.g. a kid's profile — stay subject to the quota even though the account +// holds the admin role. X-Profile-Id is client-asserted, so this is a +// budgeting gate, not a security boundary; the quota itself is account-scoped. +func (h *SubtitleAIHandler) quotaExempt(r *http.Request) bool { + ctx := r.Context() + if !apimw.IsAdmin(ctx) { + return false + } + profileID := apimw.GetProfileID(ctx) + if profileID == "" { + profileID = r.Header.Get("X-Profile-Id") + } + if profileID == "" || h.StoreProvider == nil { + return true + } + store, err := h.StoreProvider.ForUser(ctx, apimw.GetUserID(ctx)) + if err != nil { + return false // fail closed: the quota still applies + } + profile, err := store.GetProfile(ctx, profileID) + if err != nil || profile == nil { + return false + } + return profile.IsPrimary +} + +// quotaExceededMessage turns a quota error into a user-facing message with the +// limit and window when available. +func quotaExceededMessage(err error) string { + var qe *ai.QuotaExceededError + if errors.As(err, &qe) { + return fmt.Sprintf( + "You've reached your transcription limit (%d per %s). Try again later or ask an admin to raise it.", + qe.Limit, qe.Period) + } + return "You've reached your transcription limit. Try again later or ask an admin to raise it." +} + +// HandleQuota reports the calling user's transcription quota usage, so the +// player can show how many jobs they have left before they start one. +// GET /api/v1/subtitles/ai/quota +func (h *SubtitleAIHandler) HandleQuota(w http.ResponseWriter, r *http.Request) { + quota, err := h.service.TranscribeQuota(r.Context(), apimw.GetUserID(r.Context()), h.quotaExempt(r)) + if err != nil { + slog.Error("failed to load transcription quota", "error", err) + writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load transcription quota") + return + } + writeJSON(w, http.StatusOK, quota) +} + // HandleGetJob returns a job's current state. GET /api/v1/subtitles/ai/jobs/{job_id} func (h *SubtitleAIHandler) HandleGetJob(w http.ResponseWriter, r *http.Request) { job, ok := h.loadAuthorizedJob(w, r) diff --git a/internal/api/middleware/auth.go b/internal/api/middleware/auth.go index 58a86f22..325e4188 100644 --- a/internal/api/middleware/auth.go +++ b/internal/api/middleware/auth.go @@ -181,6 +181,14 @@ func GetClaims(ctx context.Context) *auth.Claims { return claims } +// IsAdmin reports whether the context's authenticated user account has the +// admin role. Returns false when no claims are present. Note this is the +// account-level role; it says nothing about which household profile is active. +func IsAdmin(ctx context.Context) bool { + claims := GetClaims(ctx) + return claims != nil && claims.Role == "admin" +} + // GetUserID retrieves the user ID from the JWT claims in the context. // Returns 0 if no claims are present. func GetUserID(ctx context.Context) int { diff --git a/internal/api/router.go b/internal/api/router.go index f1906fe2..a520e195 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -194,6 +194,10 @@ func NewRouter(deps Dependencies) chi.Router { r.Use(middleware.Recoverer) r.Use(apimw.Metrics) + // Compress text-like responses (JSON, SVG, …); media content types are + // not in the middleware's allowlist and stream through untouched. + r.Use(middleware.Compress(5)) + // Activity logging (before auth — captures all requests including failed auth). if deps.ActivityLogWriter != nil { r.Use(activitylog.NewMiddleware(deps.ActivityLogWriter, deps.NodeID)) @@ -944,13 +948,15 @@ func NewRouter(deps Dependencies) chi.Router { transcribeEnabled = false } aiCfg := subtitleai.Config{ - Configured: deps.Config.AI.BaseURL != "", - TranslateEnabled: deps.Config.SubtitleAI.Enabled, - TranscribeEnabled: transcribeEnabled, - ChatModel: deps.Config.AI.ChatModel, - ASRModel: deps.Config.AI.ASRModel, - BatchSize: deps.Config.SubtitleAI.BatchSize, - ContextNeighbors: deps.Config.SubtitleAI.ContextNeighbors, + Configured: deps.Config.AI.BaseURL != "", + TranslateEnabled: deps.Config.SubtitleAI.Enabled, + TranscribeEnabled: transcribeEnabled, + ChatModel: deps.Config.AI.ChatModel, + ASRModel: deps.Config.AI.ASRModel, + BatchSize: deps.Config.SubtitleAI.BatchSize, + ContextNeighbors: deps.Config.SubtitleAI.ContextNeighbors, + TranscribeQuotaJobs: deps.Config.SubtitleAI.TranscribeQuotaJobs, + TranscribeQuotaPeriod: deps.Config.SubtitleAI.TranscribeQuotaPeriod, } var aiNotifier subtitleai.Notifier if subtitleAINotifier != nil { @@ -972,6 +978,7 @@ func NewRouter(deps Dependencies) chi.Router { ) aiService.Recover() subtitleAIHandler = handlers.NewSubtitleAIHandler(aiService) + subtitleAIHandler.StoreProvider = deps.UserStoreProvider } // Metadata AI translation (descriptions into the localization tables). @@ -1862,6 +1869,7 @@ func NewRouter(deps Dependencies) chi.Router { r.Post("/detect-language", subtitleSearchHandler.HandleDetectLanguage) if subtitleAIHandler != nil { r.Get("/ai/status", subtitleAIHandler.HandleStatus) + r.Get("/ai/quota", subtitleAIHandler.HandleQuota) r.Post("/ai/translate", subtitleAIHandler.HandleTranslate) r.Get("/ai/jobs", subtitleAIHandler.HandleListJobs) r.Get("/ai/jobs/{job_id}", subtitleAIHandler.HandleGetJob) diff --git a/internal/config/config.go b/internal/config/config.go index f7824808..424eed76 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -273,6 +273,11 @@ type SubtitleAIConfig struct { // (60..600). Shorter chunks bound Whisper timestamp drift on long files; // longer chunks mean fewer requests and fewer boundary word-clips. ASRChunkSeconds int `yaml:"-"` + // TranscribeQuotaJobs caps how many transcription jobs each non-admin user + // may start per rolling quota period (0 = unlimited). + TranscribeQuotaJobs int `yaml:"-"` + // TranscribeQuotaPeriod is the rolling quota window: "day", "week", "month". + TranscribeQuotaPeriod string `yaml:"-"` } // MetadataAIConfig holds the metadata translation feature toggles. The diff --git a/internal/config/db_loader.go b/internal/config/db_loader.go index fcbd682c..fd385b09 100644 --- a/internal/config/db_loader.go +++ b/internal/config/db_loader.go @@ -7,6 +7,8 @@ import ( "time" "github.com/google/uuid" + + subtitleai "github.com/Silo-Server/silo-server/internal/subtitles/ai" ) // stringOr returns the value from the map for the given key, or the fallback if absent/empty. @@ -470,6 +472,21 @@ func LoadFromDB(m map[string]string) (*Config, error) { return nil, err } cfg.SubtitleAI.ASRChunkSeconds = subtitleAIChunkSeconds + // A bad quota row must not block startup: fall back to "no quota" / + // the daily window with a warning instead of failing the config load. + transcribeQuotaJobs, err := intOr(m, "subtitle_ai.transcribe_quota_jobs", 0) + if err != nil || transcribeQuotaJobs < 0 { + slog.Warn("invalid subtitle_ai.transcribe_quota_jobs setting; quota disabled", + "value", m["subtitle_ai.transcribe_quota_jobs"], "error", err) + transcribeQuotaJobs = 0 + } + cfg.SubtitleAI.TranscribeQuotaJobs = transcribeQuotaJobs + period := stringOr(m, "subtitle_ai.transcribe_quota_period", subtitleai.QuotaPeriodDay) + if !subtitleai.ValidQuotaPeriod(period) { + slog.Warn("invalid subtitle_ai.transcribe_quota_period setting; using day", "value", period) + period = subtitleai.QuotaPeriodDay + } + cfg.SubtitleAI.TranscribeQuotaPeriod = period // Metadata AI translation feature toggles. metadataAIEnabled, err := boolOr(m, "metadata_ai.enabled", false) diff --git a/internal/subtitles/ai/config.go b/internal/subtitles/ai/config.go index fec748f8..f384084b 100644 --- a/internal/subtitles/ai/config.go +++ b/internal/subtitles/ai/config.go @@ -17,6 +17,13 @@ type Config struct { ASRModel string // audio-transcription model used for ASR BatchSize int // cues per translation request ContextNeighbors int // preceding source cues sent as untranslated context + // TranscribeQuotaJobs caps how many transcription jobs (transcribe / + // transcribe_translate) each non-admin user may start per rolling quota + // period; 0 means unlimited. + TranscribeQuotaJobs int + // TranscribeQuotaPeriod is the rolling window the quota counts against: + // "day", "week", or "month". + TranscribeQuotaPeriod string } // TranslateReady reports whether subtitle translation can currently run. diff --git a/internal/subtitles/ai/job.go b/internal/subtitles/ai/job.go index 4aa1eb1b..9c793efa 100644 --- a/internal/subtitles/ai/job.go +++ b/internal/subtitles/ai/job.go @@ -78,6 +78,10 @@ type JobRequest struct { SourceLanguage string TargetLanguage string RequestedBy *int + // QuotaExempt skips the per-user transcription quota. The HTTP layer + // decides exemption policy (the household parent of an admin account + // manages the budget, so it is not bound by it); the service enforces. + QuotaExempt bool // SessionID, when set, streams live cues to that playback session. SessionID string // StartPosition (seconds) makes translation start at the viewer's playhead. diff --git a/internal/subtitles/ai/pgrepo.go b/internal/subtitles/ai/pgrepo.go index 5bbb4d6b..decf0afa 100644 --- a/internal/subtitles/ai/pgrepo.go +++ b/internal/subtitles/ai/pgrepo.go @@ -37,14 +37,69 @@ func scanJob(row pgx.Row) (*Job, error) { return &j, nil } -func (r *PgJobRepository) InsertJob(ctx context.Context, job *Job) error { +// transcribeQuotaCountSQL counts a user's quota-consuming transcription jobs. +// Failed/cancelled rows with zero progress never did ASR work (engine +// unreachable, bad media file, cancelled before start) and are excluded so +// server-side faults don't lock the user out; terminal rows with progress > 0 +// consumed compute and still count. Shared by the quota status endpoint and +// the insert-time guard so the number shown always matches the one enforced. +const transcribeQuotaCountSQL = `SELECT count(*) FROM subtitle_ai_jobs + WHERE requested_by = $1 AND created_at >= $2 + AND kind IN ('transcribe', 'transcribe_translate') + AND NOT (status IN ('failed', 'cancelled') AND progress = 0)` + +// transcribeQuotaLockNamespace partitions advisory locks so the transcription +// quota's per-user lock cannot collide with advisory locks held elsewhere in +// the database (media requests use 139, autoscan 900173). The value is +// arbitrary; what matters is that it is stable. +const transcribeQuotaLockNamespace = 412 + +func (r *PgJobRepository) InsertJob(ctx context.Context, job *Job, quota *JobQuota) error { if job.Engine == "" { job.Engine = "openai" } if job.Status == "" { job.Status = JobStatusPending } - return r.pool.QueryRow(ctx, + if quota == nil { + return r.insertJob(ctx, r.pool, job) + } + + tx, err := r.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin insert subtitle ai job: %w", err) + } + defer tx.Rollback(ctx) //nolint:errcheck // no-op after commit + + // Serialize check-and-insert per user so concurrent requests cannot race + // past the limit (same pattern as the media-request quota). + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1::int4, $2::int4)`, + transcribeQuotaLockNamespace, quota.UserID); err != nil { + return fmt.Errorf("acquire transcription quota lock: %w", err) + } + var used int + if err := tx.QueryRow(ctx, transcribeQuotaCountSQL, quota.UserID, quota.Since).Scan(&used); err != nil { + return fmt.Errorf("count transcribe jobs for quota: %w", err) + } + if used >= quota.Limit { + return &QuotaExceededError{Limit: quota.Limit, Used: used, Period: quota.Period} + } + if err := r.insertJob(ctx, tx, job); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit insert subtitle ai job: %w", err) + } + return nil +} + +// rowQuerier abstracts pool vs transaction for insertJob. +type rowQuerier interface { + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + +func (r *PgJobRepository) insertJob(ctx context.Context, q rowQuerier, job *Job) error { + return q.QueryRow(ctx, `INSERT INTO subtitle_ai_jobs (media_file_id, kind, source_index, source_language, target_language, engine, model, status, progress, progress_message, idempotency_key, requested_by) @@ -101,6 +156,15 @@ func (r *PgJobRepository) ListJobsByMediaFile(ctx context.Context, mediaFileID i return jobs, rows.Err() } +func (r *PgJobRepository) CountTranscribeJobsByUserSince(ctx context.Context, userID int, since time.Time) (int, error) { + var count int + err := r.pool.QueryRow(ctx, transcribeQuotaCountSQL, userID, since).Scan(&count) + if err != nil { + return 0, fmt.Errorf("count transcribe jobs by user: %w", err) + } + return count, nil +} + // UpdateProgress, CompleteJob, and FailJob only transition a job that is still // active ("pending"/"running"). The guard makes them no-ops on an already // terminal row, so a job that was cancelled or reaped as stale can never be diff --git a/internal/subtitles/ai/quota.go b/internal/subtitles/ai/quota.go new file mode 100644 index 00000000..14887f17 --- /dev/null +++ b/internal/subtitles/ai/quota.go @@ -0,0 +1,127 @@ +package ai + +import ( + "context" + "errors" + "fmt" + "time" +) + +// ErrQuotaExceeded is returned when a transcription request would exceed the +// per-user job quota, so handlers can map it to 429. +var ErrQuotaExceeded = errors.New("transcription quota exceeded") + +// Quota period names, stored in subtitle_ai.transcribe_quota_period. Periods +// are rolling windows ending now, not calendar buckets, so there is no reset +// moment to game. +const ( + QuotaPeriodDay = "day" + QuotaPeriodWeek = "week" + QuotaPeriodMonth = "month" +) + +// ValidQuotaPeriod reports whether period names a supported quota window. +// The settings write path and the config loader both validate against this, +// so the period vocabulary lives in one place. +func ValidQuotaPeriod(period string) bool { + switch period { + case QuotaPeriodDay, QuotaPeriodWeek, QuotaPeriodMonth: + return true + } + return false +} + +// QuotaPeriodWindow maps a period name to its rolling window length. Unknown +// values fall back to a day, matching the config loader's default. +func QuotaPeriodWindow(period string) time.Duration { + switch period { + case QuotaPeriodWeek: + return 7 * 24 * time.Hour + case QuotaPeriodMonth: + return 30 * 24 * time.Hour + default: + return 24 * time.Hour + } +} + +// JobQuota is the per-user cap InsertJob enforces atomically with the insert, +// so concurrent requests cannot race past the limit. A nil *JobQuota means +// the insert is unconditional. +type JobQuota struct { + UserID int + Limit int + Since time.Time + Period string // window name, carried into QuotaExceededError for messages +} + +// QuotaExceededError carries the usage details behind ErrQuotaExceeded. +type QuotaExceededError struct { + Limit int + Used int + Period string +} + +func (e *QuotaExceededError) Error() string { + return fmt.Sprintf("transcription limit reached: %d jobs per %s", e.Limit, e.Period) +} + +func (e *QuotaExceededError) Unwrap() error { return ErrQuotaExceeded } + +// QuotaStatus reports a user's transcription quota usage. Limited is false +// when no quota applies to the caller (quota disabled, or an exempt +// requester); the remaining fields are only meaningful when it is true. +type QuotaStatus struct { + Limited bool `json:"limited"` + Limit int `json:"limit"` + Used int `json:"used"` + Remaining int `json:"remaining"` + Period string `json:"period"` +} + +// TranscribeQuota returns the transcription quota status for a user. Exempt +// callers always get an unlimited status. The count uses the same query the +// insert-time enforcement runs, so the number shown always matches the +// number enforced. +func (s *Service) TranscribeQuota(ctx context.Context, userID int, exempt bool) (QuotaStatus, error) { + if exempt || s.cfg.TranscribeQuotaJobs <= 0 { + return QuotaStatus{}, nil + } + used, err := s.repo.CountTranscribeJobsByUserSince(ctx, userID, s.quotaWindowStart()) + if err != nil { + return QuotaStatus{}, err + } + remaining := s.cfg.TranscribeQuotaJobs - used + if remaining < 0 { + remaining = 0 + } + return QuotaStatus{ + Limited: true, + Limit: s.cfg.TranscribeQuotaJobs, + Used: used, + Remaining: remaining, + Period: s.cfg.TranscribeQuotaPeriod, + }, nil +} + +// transcribeQuotaSpec returns the quota InsertJob must enforce for this +// request, or nil when none applies: translation-only jobs, exempt +// requesters, requests without an attributable user (internal callers), and +// servers with the quota disabled all pass through. +func (s *Service) transcribeQuotaSpec(req JobRequest) *JobQuota { + if !req.Kind.IsTranscribe() || req.QuotaExempt || req.RequestedBy == nil { + return nil + } + if s.cfg.TranscribeQuotaJobs <= 0 { + return nil + } + return &JobQuota{ + UserID: *req.RequestedBy, + Limit: s.cfg.TranscribeQuotaJobs, + Since: s.quotaWindowStart(), + Period: s.cfg.TranscribeQuotaPeriod, + } +} + +func (s *Service) quotaWindowStart() time.Time { + return time.Now().Add(-QuotaPeriodWindow(s.cfg.TranscribeQuotaPeriod)) +} diff --git a/internal/subtitles/ai/repo.go b/internal/subtitles/ai/repo.go index 7ae8d192..445e716f 100644 --- a/internal/subtitles/ai/repo.go +++ b/internal/subtitles/ai/repo.go @@ -7,12 +7,22 @@ import ( // JobRepository persists AI subtitle jobs. type JobRepository interface { - InsertJob(ctx context.Context, job *Job) error + // InsertJob inserts a new job. When quota is non-nil it enforces the + // per-user transcription cap atomically with the insert (serialized per + // user), returning a *QuotaExceededError instead of inserting once the + // user has quota.Limit countable jobs since quota.Since. + InsertJob(ctx context.Context, job *Job, quota *JobQuota) error GetJob(ctx context.Context, id int64) (*Job, error) // GetActiveJobByIdempotencyKey returns a pending/running job with the given // key, or nil if none exists. GetActiveJobByIdempotencyKey(ctx context.Context, key string) (*Job, error) ListJobsByMediaFile(ctx context.Context, mediaFileID int) ([]Job, error) + // CountTranscribeJobsByUserSince counts the user's quota-consuming + // transcription jobs (transcribe / transcribe_translate) created at or + // after `since`. Failed or cancelled jobs that never produced any + // transcription work are excluded, so server-side faults don't lock the + // user out; jobs that burned ASR compute count even if they later failed. + CountTranscribeJobsByUserSince(ctx context.Context, userID int, since time.Time) (int, error) UpdateProgress(ctx context.Context, id int64, status JobStatus, progress float64, message string) error CompleteJob(ctx context.Context, id int64, subtitleID int) error FailJob(ctx context.Context, id int64, status JobStatus, message string) error diff --git a/internal/subtitles/ai/service.go b/internal/subtitles/ai/service.go index d3b0e55e..10bc9dab 100644 --- a/internal/subtitles/ai/service.go +++ b/internal/subtitles/ai/service.go @@ -171,6 +171,12 @@ func (s *Service) Enqueue(ctx context.Context, req JobRequest) (*Job, error) { return existing, nil } + // Quota gates only the creation of a new job: collapsing onto an existing + // in-flight job above costs nothing, so it stays allowed at the limit. + // Enforcement happens inside InsertJob, atomically with the insert, so + // concurrent requests cannot race past the limit. + quota := s.transcribeQuotaSpec(req) + job := &Job{ MediaFileID: req.MediaFileID, Kind: req.Kind, @@ -186,8 +192,10 @@ func (s *Service) Enqueue(ctx context.Context, req JobRequest) (*Job, error) { SessionID: req.SessionID, StartPosition: req.StartPosition, } - if err := s.repo.InsertJob(ctx, job); err != nil { + if err := s.repo.InsertJob(ctx, job, quota); err != nil { // A racing duplicate trips the partial unique index; return the winner. + // The same lookup also covers a quota rejection that raced an identical + // request: collapsing onto the winner is free, so it stays allowed. if existing, lookupErr := s.repo.GetActiveJobByIdempotencyKey(ctx, key); lookupErr == nil && existing != nil { return existing, nil } diff --git a/internal/subtitles/ai/service_test.go b/internal/subtitles/ai/service_test.go index 127ef9e1..ab1ae961 100644 --- a/internal/subtitles/ai/service_test.go +++ b/internal/subtitles/ai/service_test.go @@ -2,27 +2,49 @@ package ai import ( "context" + "errors" "sync" "testing" "time" "github.com/Silo-Server/silo-server/internal/ai/jobrunner" + "github.com/Silo-Server/silo-server/internal/models" ) -// recordingRepo is a JobRepository that records ResetStaleJobs calls and no-ops -// everything else, so the recovery/reaper behavior can be tested in isolation. +// recordingRepo is a JobRepository that records ResetStaleJobs calls and +// insert/quota-count activity and no-ops everything else, so service behavior +// can be tested in isolation. type recordingRepo struct { mu sync.Mutex resets int lastBefore time.Time + inserts int + quotaUsed int // returned by CountTranscribeJobsByUserSince } -func (r *recordingRepo) InsertJob(context.Context, *Job) error { return nil } +// InsertJob mirrors the Postgres repo's atomic quota guard: a non-nil quota +// is checked against quotaUsed before the insert, so the tests exercise the +// same enforcement path production uses. +func (r *recordingRepo) InsertJob(_ context.Context, job *Job, quota *JobQuota) error { + r.mu.Lock() + defer r.mu.Unlock() + if quota != nil && r.quotaUsed >= quota.Limit { + return &QuotaExceededError{Limit: quota.Limit, Used: r.quotaUsed, Period: quota.Period} + } + r.inserts++ + job.ID = int64(r.inserts) + return nil +} func (r *recordingRepo) GetJob(context.Context, int64) (*Job, error) { return nil, nil } func (r *recordingRepo) GetActiveJobByIdempotencyKey(context.Context, string) (*Job, error) { return nil, nil } func (r *recordingRepo) ListJobsByMediaFile(context.Context, int) ([]Job, error) { return nil, nil } +func (r *recordingRepo) CountTranscribeJobsByUserSince(context.Context, int, time.Time) (int, error) { + r.mu.Lock() + defer r.mu.Unlock() + return r.quotaUsed, nil +} func (r *recordingRepo) UpdateProgress(context.Context, int64, JobStatus, float64, string) error { return nil } @@ -44,6 +66,121 @@ func (r *recordingRepo) snapshot() (int, time.Time) { return r.resets, r.lastBefore } +// stubTranscriber satisfies Transcriber so TranscribeEnabled() is true in +// tests; it returns no cues. +type stubTranscriber struct{} + +func (stubTranscriber) Transcribe(context.Context, TranscribeJobRequest, + func([]SubtitleCue, int, int)) ([]SubtitleCue, string, error) { + return nil, "", nil +} + +// nilFileResolver reports every media file as missing, so dispatched test jobs +// fail fast instead of touching the filesystem. +type nilFileResolver struct{} + +func (nilFileResolver) GetByID(context.Context, int) (*models.MediaFile, error) { return nil, nil } + +// newQuotaTestService builds a transcribe-ready service whose repo reports +// `used` transcription jobs in the current quota window. +func newQuotaTestService(t *testing.T, used, limit int) (*Service, *recordingRepo) { + t.Helper() + repo := &recordingRepo{quotaUsed: used} + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + cfg := Config{ + Configured: true, + TranscribeEnabled: true, + ASRModel: "whisper-1", + TranscribeQuotaJobs: limit, + TranscribeQuotaPeriod: QuotaPeriodDay, + } + svc := NewService(ctx, cfg, repo, nil, stubTranscriber{}, nil, nil, nilFileResolver{}, nil, "", nil, nil) + return svc, repo +} + +func transcribeRequest(userID int, exempt bool) JobRequest { + return JobRequest{ + MediaFileID: 1, + Kind: JobKindTranscribe, + SourceIndex: -1, + RequestedBy: &userID, + QuotaExempt: exempt, + } +} + +func TestEnqueueRejectsTranscribeOverQuota(t *testing.T) { + svc, repo := newQuotaTestService(t, 3, 3) + + _, err := svc.Enqueue(context.Background(), transcribeRequest(42, false)) + if !errors.Is(err, ErrQuotaExceeded) { + t.Fatalf("Enqueue error = %v, want ErrQuotaExceeded", err) + } + var qe *QuotaExceededError + if !errors.As(err, &qe) { + t.Fatalf("error %v does not unwrap to *QuotaExceededError", err) + } + if qe.Limit != 3 || qe.Used != 3 || qe.Period != QuotaPeriodDay { + t.Errorf("quota error = %+v, want limit=3 used=3 period=day", qe) + } + if repo.inserts != 0 { + t.Errorf("job was inserted despite exceeded quota (inserts=%d)", repo.inserts) + } +} + +func TestEnqueueAllowsTranscribeUnderQuota(t *testing.T) { + svc, repo := newQuotaTestService(t, 2, 3) + + if _, err := svc.Enqueue(context.Background(), transcribeRequest(42, false)); err != nil { + t.Fatalf("Enqueue under quota failed: %v", err) + } + if repo.inserts != 1 { + t.Errorf("inserts = %d, want 1", repo.inserts) + } +} + +func TestEnqueueQuotaExemptions(t *testing.T) { + t.Run("exempt", func(t *testing.T) { + svc, repo := newQuotaTestService(t, 10, 1) + if _, err := svc.Enqueue(context.Background(), transcribeRequest(42, true)); err != nil { + t.Fatalf("exempt Enqueue blocked by quota: %v", err) + } + if repo.inserts != 1 { + t.Errorf("inserts = %d, want 1", repo.inserts) + } + }) + t.Run("unlimited", func(t *testing.T) { + svc, repo := newQuotaTestService(t, 10, 0) + if _, err := svc.Enqueue(context.Background(), transcribeRequest(42, false)); err != nil { + t.Fatalf("Enqueue with quota disabled failed: %v", err) + } + if repo.inserts != 1 { + t.Errorf("inserts = %d, want 1", repo.inserts) + } + }) +} + +func TestTranscribeQuotaStatus(t *testing.T) { + svc, _ := newQuotaTestService(t, 2, 5) + + got, err := svc.TranscribeQuota(context.Background(), 42, false) + if err != nil { + t.Fatalf("TranscribeQuota: %v", err) + } + want := QuotaStatus{Limited: true, Limit: 5, Used: 2, Remaining: 3, Period: QuotaPeriodDay} + if got != want { + t.Errorf("TranscribeQuota = %+v, want %+v", got, want) + } + + exempt, err := svc.TranscribeQuota(context.Background(), 42, true) + if err != nil { + t.Fatalf("TranscribeQuota (exempt): %v", err) + } + if exempt.Limited { + t.Errorf("exempt quota = %+v, want unlimited", exempt) + } +} + // Recover reaps immediately using a heartbeat cutoff of now-staleJobThreshold, // not "every active job", so a live worker's jobs survive a peer's startup. func TestRecoverReapsStaleJobsImmediately(t *testing.T) { diff --git a/migrations/sql/20260610195318_add_subtitle_ai_jobs_quota_index.sql b/migrations/sql/20260610195318_add_subtitle_ai_jobs_quota_index.sql new file mode 100644 index 00000000..100357af --- /dev/null +++ b/migrations/sql/20260610195318_add_subtitle_ai_jobs_quota_index.sql @@ -0,0 +1,15 @@ +-- +goose Up +-- +goose StatementBegin +-- The transcription quota counts a user's recent transcription jobs on every +-- enqueue and every quota lookup (player translate modal). Job rows are never +-- deleted, so without an index that count degrades to a full sequential scan +-- as job history grows. +CREATE INDEX subtitle_ai_jobs_quota_idx + ON public.subtitle_ai_jobs (requested_by, created_at) + WHERE kind IN ('transcribe', 'transcribe_translate'); +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +DROP INDEX IF EXISTS subtitle_ai_jobs_quota_idx; +-- +goose StatementEnd diff --git a/web/src/lib/quotaPeriods.ts b/web/src/lib/quotaPeriods.ts new file mode 100644 index 00000000..8d4f2f54 --- /dev/null +++ b/web/src/lib/quotaPeriods.ts @@ -0,0 +1,10 @@ +// Rolling-window labels for the subtitle-AI transcription quota periods. +// One source of truth for every surface that names a period; keep in sync +// with QuotaPeriodWindow in internal/subtitles/ai/quota.go. +export const QUOTA_PERIODS = ["day", "week", "month"] as const; + +export const QUOTA_PERIOD_WINDOW_LABELS: Record = { + day: "24 hours", + week: "7 days", + month: "30 days", +}; diff --git a/web/src/pages/admin-settings/AIServicesSettings.tsx b/web/src/pages/admin-settings/AIServicesSettings.tsx index 3966571b..5da9bd5b 100644 --- a/web/src/pages/admin-settings/AIServicesSettings.tsx +++ b/web/src/pages/admin-settings/AIServicesSettings.tsx @@ -7,6 +7,7 @@ import { } from "@/hooks/queries/admin/settings"; import { Button } from "@/components/ui/button"; +import { QUOTA_PERIODS, QUOTA_PERIOD_WINDOW_LABELS } from "@/lib/quotaPeriods"; import { CredentialStatus } from "./CredentialStatus"; import { SettingField } from "./SettingField"; @@ -264,6 +265,8 @@ function AIFeaturesCard() { const [batchSize, setBatchSize] = useState("40"); const [contextNeighbors, setContextNeighbors] = useState("2"); const [asrChunkSeconds, setAsrChunkSeconds] = useState("600"); + const [transcribeQuotaJobs, setTranscribeQuotaJobs] = useState("0"); + const [transcribeQuotaPeriod, setTranscribeQuotaPeriod] = useState("day"); useEffect(() => { if (!settings) return; @@ -274,6 +277,8 @@ function AIFeaturesCard() { setBatchSize(settings["subtitle_ai.batch_size"] ?? "40"); setContextNeighbors(settings["subtitle_ai.context_neighbors"] ?? "2"); setAsrChunkSeconds(settings["subtitle_ai.asr_chunk_seconds"] ?? "600"); + setTranscribeQuotaJobs(settings["subtitle_ai.transcribe_quota_jobs"] ?? "0"); + setTranscribeQuotaPeriod(settings["subtitle_ai.transcribe_quota_period"] ?? "day"); }, [settings]); function save() { @@ -296,6 +301,11 @@ function AIFeaturesCard() { toast.error("Transcription chunk length must be between 60 and 600 seconds."); return; } + const parsedQuotaJobs = Number.parseInt(transcribeQuotaJobs, 10); + if (!Number.isInteger(parsedQuotaJobs) || parsedQuotaJobs < 0) { + toast.error("Transcription limit must be zero (unlimited) or a positive whole number."); + return; + } void Promise.all([ updateSetting.mutateAsync({ key: "subtitle_ai.enabled", value: subtitleTranslate }), updateSetting.mutateAsync({ key: "subtitle_ai.transcribe_enabled", value: transcribe }), @@ -310,6 +320,14 @@ function AIFeaturesCard() { key: "subtitle_ai.asr_chunk_seconds", value: String(parsedChunkSeconds), }), + updateSetting.mutateAsync({ + key: "subtitle_ai.transcribe_quota_jobs", + value: String(parsedQuotaJobs), + }), + updateSetting.mutateAsync({ + key: "subtitle_ai.transcribe_quota_period", + value: transcribeQuotaPeriod, + }), ]); } @@ -376,6 +394,24 @@ function AIFeaturesCard() { onChange={setAsrChunkSeconds} hint="60–600. Shorter chunks keep Whisper timestamps tighter on long files (try 300 if subtitles drift), at the cost of more requests and occasional clipped words at chunk boundaries." /> + + ({ + value: p, + label: `Per ${p} (rolling ${QUOTA_PERIOD_WINDOW_LABELS[p]})`, + }))} + hint="Rolling window the transcription limit counts against." + />