package web import ( "context" "errors" "fmt" "io" "math" "net/http" "path/filepath" "regexp" "strings" "time" "unicode" "unicode/utf8" "github.com/a-h/templ" "git.d-ma.be/mathias/tapir/internal/adapters/chat" "git.d-ma.be/mathias/tapir/internal/adapters/store" "git.d-ma.be/mathias/tapir/internal/domain" "git.d-ma.be/mathias/tapir/internal/errkind" "git.d-ma.be/mathias/tapir/internal/uploads" ) // Uploader accepts an uploaded recording for transcription (ADR-033). // *uploads.Intake satisfies it. It is the whole of what the handlers know about // validation, spooling and queueing. type Uploader interface { Accept(ctx context.Context, userID, title string, src io.Reader) (id string, err error) } const ( // uploadOverhead is headroom above the file-size cap for the multipart framing // and the title field; the cap itself is enforced on the file part by Intake. uploadOverhead = 1 << 20 maxTitleRunes = 200 // uploadPollEvery is how often an unfinished upload re-asks for its status. uploadPollEvery = "2s" // serviceSilentAfter is how long a job may go without any signal (heartbeat or // progress) before the page calls the service lost. The heartbeat is every 10 s, so // 45 s is four missed beats (ADR-035). serviceSilentAfter = 45 * time.Second // countedQuietAfter is how long a COUNTED phase (the deciding one) may go without // advancing before the page says it is quiet. Each decision has a two-minute timeout, // so five minutes is more than two in a row. A phase with no count is never quiet. countedQuietAfter = 5 * time.Minute ) // uploadsView is what the /uploads page renders. type uploadsView struct { Limits uploads.Limits Rows []domain.UploadedTranscript Error string // a synchronous rejection message, empty when none } // handleUploadsPage renders the upload form and the caller's own uploads. func (a *App) handleUploadsPage(w http.ResponseWriter, r *http.Request) { userID, ok := a.currentUserID(w, r) if !ok { return } a.renderUploadsPage(w, r, userID, http.StatusOK, "") } func (a *App) renderUploadsPage(w http.ResponseWriter, r *http.Request, userID string, status int, msg string) { rows, err := a.Store.ListUploads(r.Context(), userID) if err != nil { a.serverError(w, r, "list uploads", err) return } a.renderStatus(w, r, status, UploadsPage(uploadsView{Limits: a.UploadLimits, Rows: rows, Error: msg})) } // handleUploadPost streams the multipart body: the optional title field, then the // file part, which goes straight to Intake without being buffered in memory. A // rejection is synchronous — a status code and a message on the upload page, no // row, no timeout (ADR-033). func (a *App) handleUploadPost(w http.ResponseWriter, r *http.Request) { userID, ok := a.currentUserID(w, r) if !ok { return } r.Body = http.MaxBytesReader(w, r.Body, a.UploadLimits.MaxBytes+uploadOverhead) mr, err := r.MultipartReader() if err != nil { a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "Choose an audio file to upload.") return } title := "" for { part, err := mr.NextPart() if errors.Is(err, io.EOF) { break } if err != nil { var tooBig *http.MaxBytesError if !errors.As(err, &tooBig) { // A malformed multipart body (a header with a control byte, a truncated // part) is the sender's to fix: a 400, not a server error, and the error // text is never logged: it quotes the headers, which carry the file name. a.logger().Info("upload rejected", "reason", "malformed") a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "That upload could not be read. Please try again.") return } a.uploadFailure(w, r, userID, err) return } switch part.FormName() { case "title": b, err := io.ReadAll(io.LimitReader(part, maxTitleRunes*utf8.UTFMax)) if err != nil { a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "That upload could not be read. Please try again.") return } title = cleanTitle(string(b)) case "file": if title == "" { title = titleFromFilename(part.FileName()) } id, err := a.Uploads.Accept(r.Context(), userID, title, part) if err != nil { a.uploadFailure(w, r, userID, err) return } http.Redirect(w, r, "/uploads/"+id, http.StatusSeeOther) return } } a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "Choose an audio file to upload.") } // uploadFailure maps an Accept (or multipart) error to a response: a validation // rejection is the user's to fix (422, or 413 for size) and carries its own // message; anything else is ours. func (a *App) uploadFailure(w http.ResponseWriter, r *http.Request, userID string, err error) { var rej *uploads.RejectError var tooBig *http.MaxBytesError switch { case errors.As(err, &rej): // Logged by reason only: the filename and title are private content. a.logger().Info("upload rejected", "reason", string(rej.Reason)) status := http.StatusUnprocessableEntity switch rej.Reason { case uploads.ReasonTooLarge: status = http.StatusRequestEntityTooLarge case uploads.ReasonBusy: status = http.StatusTooManyRequests } a.renderUploadsPage(w, r, userID, status, rej.Message) case errors.As(err, &tooBig): a.logger().Info("upload rejected", "reason", string(uploads.ReasonTooLarge)) a.renderUploadsPage(w, r, userID, http.StatusRequestEntityTooLarge, "This file is larger than the upload limit.") case errors.Is(err, io.ErrUnexpectedEOF): // The sender hung up mid-file: theirs to retry, not a server fault. a.logger().Info("upload rejected", "reason", "truncated") a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "The upload was cut off before it finished. Please try again.") default: // Not err: it can carry the file's name, a spool path or a database value. a.logger().Error("upload failed", "kind", errkind.Of(err)) http.Error(w, "internal error", http.StatusInternalServerError) } } // handleUploadDetail renders one upload's page (GET /uploads/{id}). The row is // read through the user-scoped store, so another user's id is a 404 — existence is // not disclosed, and the title never appears in the response. func (a *App) handleUploadDetail(w http.ResponseWriter, r *http.Request) { u, ok := a.loadOwnedUpload(w, r) if !ok { return } a.render(w, r, UploadPage(*u, a.Chat != nil, a.now())) } // handleUploadStatus renders the status fragment HTMX polls. Unfinished states // keep polling; terminal states render without hx-trigger so the page stops // asking (ADR-025: never a stuck spinner). func (a *App) handleUploadStatus(w http.ResponseWriter, r *http.Request) { u, ok := a.loadOwnedUpload(w, r) if !ok { return } a.render(w, r, uploadStatus(*u, a.Chat != nil, a.now())) } func (a *App) loadOwnedUpload(w http.ResponseWriter, r *http.Request) (*domain.UploadedTranscript, bool) { userID, ok := a.currentUserID(w, r) if !ok { return nil, false } id := r.PathValue("id") if !uuidShape.MatchString(id) { // Not an id at all: the same 404 as an unknown or foreign one, never a database // error that surfaces as a 500 (and no oracle about which ids exist). http.NotFound(w, r) return nil, false } u, err := a.Store.GetUpload(r.Context(), userID, id) if errors.Is(err, store.ErrUploadNotFound) { http.NotFound(w, r) return nil, false } if err != nil { a.serverError(w, r, "get upload", err) return nil, false } return u, true } // titleFromFilename is the default title: the file's name without its extension. func titleFromFilename(name string) string { base := filepath.Base(name) t := cleanTitle(strings.TrimSuffix(base, filepath.Ext(base))) if t == "" || t == "." { return "Untitled recording" } return t } // uuidShape is the form of an upload id. Anything else cannot be one, so it is a 404. var uuidShape = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`) // isBidiControl reports the invisible characters that change text direction. Emoji joiners // (ZWJ, ZWNJ) and variation selectors are not among them and are kept. func isBidiControl(r rune) bool { return r >= 0x202A && r <= 0x202E || r >= 0x2066 && r <= 0x2069 || r == 0x200E || r == 0x200F || r == 0x061C } // cleanTitle makes a title safe to store and show: invalid UTF-8 dropped, control // characters (NUL, newlines, escapes) turned into spaces, runs of space collapsed, trimmed // and clipped. A title is cosmetic; bad bytes in it must never fail an upload, and Postgres // refuses NUL in text. func cleanTitle(s string) string { s = strings.ToValidUTF8(s, "") s = strings.Map(func(r rune) rune { switch { case isBidiControl(r): return -1 // dropped: U+202E would reverse the rest of the title on its owner's page case unicode.IsControl(r): return ' ' } return r }, s) return clipRunes(strings.Join(strings.Fields(s), " "), maxTitleRunes) } func clipRunes(s string, n int) string { if utf8.RuneCountInString(s) <= n { return s } return string([]rune(s)[:n]) } // uploadFailureMessage turns a typed failure kind into words for the user. It is // keyed on OUR kinds, never on a provider's error string (ADR-033). func uploadFailureMessage(k domain.UploadErrorKind) string { switch k { case domain.UploadErrInterrupted: return "The server restarted before this recording was transcribed. Please upload it again." case domain.UploadErrBusy: return "Too many recordings were waiting to be transcribed when this one was uploaded. Please upload it again in a few minutes." case domain.UploadErrTooLong: return "This recording is longer than the limit once its audio is read (the file's header said otherwise). Please upload a shorter recording." case domain.UploadErrTranscriptionFailed: return "Transcription failed on every available service. Please upload it again later." default: return "This upload failed. Please upload it again." } } // handleUploadChat renders the chat page for a FINISHED upload (GET). The row is // read through the user-scoped store, so another user's id is a 404 before any // model is involved. An unfinished upload has no transcript to ask about yet, so // it goes back to its status page. func (a *App) handleUploadChat(w http.ResponseWriter, r *http.Request) { u, ok := a.loadOwnedUpload(w, r) if !ok { return } if u.Status != domain.UploadDone { http.Redirect(w, r, "/uploads/"+u.ID, http.StatusSeeOther) return } model, _ := a.resolveUploadModel("") a.render(w, r, UploadChatPage(*u, a.uploadChatView(*u, model, nil, false, ""))) } // handleUploadChatMessage answers one question against the owner's uploaded // transcript (POST), reusing the ADR-027 Chatter. Nothing is persisted: the prior // turns ride in the form. The transcript goes to the model with the UPLOAD budget, // not the caption default. A model failure is inline, not a 500. func (a *App) handleUploadChatMessage(w http.ResponseWriter, r *http.Request) { u, ok := a.loadOwnedUpload(w, r) if !ok { return } if u.Status != domain.UploadDone { http.Redirect(w, r, "/uploads/"+u.ID, http.StatusSeeOther) return } if err := r.ParseForm(); err != nil { http.Error(w, "bad form", http.StatusBadRequest) return } model, ok := a.resolveUploadModel(r.FormValue("model")) if !ok { a.logger().Error("upload chat: no local model is configured; nothing was sent", "upload", u.ID) view := a.uploadChatView(*u, "", nil, false, "No local model is available for chat about a recording.") a.render(w, r, UploadChatPage(*u, view)) return } history := parseHistory(r.Form["hq"], r.Form["ha"]) question := strings.TrimSpace(r.FormValue("question")) view := a.uploadChatView(*u, model, history, false, "") if question != "" { reply, err := a.Chat.Answer(r.Context(), chat.Request{ Model: model, Transcript: u.Content, History: history, Question: question, MaxChars: a.UploadChatMaxChars, }) if err != nil { a.logger().Error("upload chat answer", "upload", u.ID, "model", model, "kind", errkind.Of(err)) view.Error = "That model couldn't answer just now. Try again, or switch models." } else { view.History = appendTurn(history, chat.Turn{Question: question, Answer: reply.Answer}) view.Truncated = reply.Truncated } } if isHTMX(r) { a.render(w, r, chatPanel(view)) return } a.render(w, r, UploadChatPage(*u, view)) } func (a *App) uploadChatView(u domain.UploadedTranscript, model string, history []chat.Turn, truncated bool, errMsg string) chatView { return chatView{ VideoID: u.ID, Title: u.Title, Available: true, Models: a.UploadChatModels, Selected: model, History: history, Truncated: truncated, Error: errMsg, PostURL: templ.SafeURL("/uploads/" + u.ID + "/chat"), Subject: "recording", } } // maxSpotsShown is how many spots to check the page lists. The stored list holds every // judged spot and the counts are exact; a page of two hundred "check this" items is not a // page anyone reads, so the rest are counted, not listed. const maxSpotsShown = 50 // minuteMark renders where in the recording a spot is, as the nearest minute. func minuteMark(seconds float64) string { return fmt.Sprintf("~%d min", int(math.Round(seconds/60))) } // shownSpots is the part of the stored list the page renders: the spots a reader should // check, changed or uncertain, in order, capped. A spot the model confidently kept is // stored (ADR-035) but is not one to check, so it is not listed. func shownSpots(r *domain.Reconciliation) []domain.ContestedSpot { var out []domain.ContestedSpot for _, c := range r.Contested { if c.Used == "alt" || c.Unsure { out = append(out, c) if len(out) == maxSpotsShown { break } } } return out } // moreSpots is how many spots to check are not listed, from the exact counts. func moreSpots(r *domain.Reconciliation) int { return max(0, r.Changed+r.Unsure-len(shownSpots(r))) } // jobHealth is what the page can honestly say about a job it is still polling. type jobHealth string const ( // healthRunning: the service is reporting and work is moving, or there is nothing yet // to say otherwise. healthRunning jobHealth = "running" // healthQuiet: the service is alive but a counted phase has not advanced for a while. healthQuiet jobHealth = "quiet" // healthLost: nothing has been heard from the service for too long. It may have // restarted or crashed; the start-up sweep fails the row once it is back. healthLost jobHealth = "lost" ) // uploadHealth tells running from quiet from lost for an upload being transcribed (the // page calls it for no other state: a queued or finished job has no heartbeat to judge). The // two timestamps mean different things (ADR-035): alive_at is the heartbeat, which says // the service is still reporting; progress_at moves only when work really moved. A // single blocking engine call has no sub-progress, so only a counted phase can be quiet. // silent is how long since the signal that decided it. func uploadHealth(u domain.UploadedTranscript, now time.Time) (health jobHealth, silent time.Duration) { last := u.AliveAt if u.ProgressAt.After(last) { last = u.ProgressAt } if last.IsZero() { return healthRunning, 0 } if d := now.Sub(last); d > serviceSilentAfter { return healthLost, d } for _, s := range u.Progress { if s.State == domain.StepRunning && s.Total > 0 && !u.ProgressAt.IsZero() { if d := now.Sub(u.ProgressAt); d > countedQuietAfter { return healthQuiet, d } } } return healthRunning, 0 } // stepDuration renders how long a phase took or has run: "<1s", "42s", "3m 12s", "1h 02m". func stepDuration(d time.Duration) string { switch { case d < time.Second: return "<1s" case d < time.Minute: return fmt.Sprintf("%ds", int(d.Seconds())) case d < time.Hour: return fmt.Sprintf("%dm %02ds", int(d.Minutes()), int(d.Seconds())%60) default: return fmt.Sprintf("%dh %02dm", int(d.Hours()), int(d.Minutes())%60) } } // stepTime is the time shown beside a phase: how long it has run, or how long it took. func stepTime(s domain.ProgressStep, now time.Time) string { switch s.State { case domain.StepRunning: return stepDuration(now.Sub(s.StartedAt)) case domain.StepDone, domain.StepFailed: return stepDuration(s.FinishedAt.Sub(s.StartedAt)) } return "" } // stepMark is the glyph at the head of a phase line. func stepMark(s domain.ProgressStep) string { switch s.State { case domain.StepDone: return "✓" case domain.StepFailed: return "✗" case domain.StepSkipped: return "–" } return "·" } // minutesAgo is a whole number of minutes for the health notes: "6 min", or "under a // minute" when it is less. func minutesAgo(d time.Duration) string { if m := int(d.Minutes()); m >= 1 { return fmt.Sprintf("%d min", m) } return "under a minute" } // resolveUploadModel picks the model for chat over an uploaded recording from the // offered LOCAL set only: the posted one when it is in the set, else the configured // default when it is, else the first. ok is false when no model is offered, in which // case nothing may be sent anywhere (the whole private transcript is the payload). func (a *App) resolveUploadModel(posted string) (model string, ok bool) { offered := func(m string) bool { for _, o := range a.UploadChatModels { if o == m { return true } } return false } switch { case posted != "" && offered(posted): return posted, true case a.UploadChatModel != "" && offered(a.UploadChatModel): return a.UploadChatModel, true case len(a.UploadChatModels) > 0: return a.UploadChatModels[0], true } return "", false }