待ち行列への登録と状態表示を追加（新着からの登録、/queue、DATABASE_URL）

2026-10-03 19:07:53 +09:00
parent a71b915c81
commit 4a02526e76
20 changed files with 1146 additions and 14 deletions
@@ -0,0 +1,191 @@
+package web
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+
+	"gitea.kazu634.com/kazu634/tver/internal/store"
+	"github.com/jackc/pgx/v5"
+)
+
+const queueListLimit = 50
+
+type queueJobView struct {
+	store.Job
+	SourceText, CreatedAtText, PhaseText string
+	PercentText, SpeedText, ETAText      string
+	FragmentText, ErrorKindText          string
+}
+
+type queueView struct {
+	Running, Queued, Failed, Canceled, Succeeded []queueJobView
+	Poll                                         bool
+}
+
+func (s *server) queuePage(w http.ResponseWriter, r *http.Request) {
+	view, err := s.loadQueue(r.Context())
+	if err != nil {
+		s.logger.Printf("待ち行列の取得に失敗: %v", err)
+		http.Error(w, "待ち行列を取得できませんでした", http.StatusInternalServerError)
+		return
+	}
+	s.render(w, "layout", page{Title: "待ち行列", Queue: &view})
+}
+
+func (s *server) queueFragment(w http.ResponseWriter, r *http.Request) {
+	view, err := s.loadQueue(r.Context())
+	if err != nil {
+		s.logger.Printf("待ち行列の取得に失敗: %v", err)
+		http.Error(w, "待ち行列を取得できませんでした", http.StatusInternalServerError)
+		return
+	}
+	s.render(w, "queue_list", view)
+}
+
+func (s *server) loadQueue(ctx context.Context) (queueView, error) {
+	var view queueView
+	for _, item := range []struct {
+		status string
+		assign func([]queueJobView)
+	}{
+		{"running", func(j []queueJobView) { view.Running = j }},
+		{"queued", func(j []queueJobView) { view.Queued = j }},
+		{"failed", func(j []queueJobView) { view.Failed = j }},
+		{"canceled", func(j []queueJobView) { view.Canceled = j }},
+		{"succeeded", func(j []queueJobView) { view.Succeeded = j }},
+	} {
+		jobs, err := s.jobs.ListJobs(ctx, item.status, queueListLimit)
+		if err != nil {
+			return queueView{}, err
+		}
+		views := make([]queueJobView, 0, len(jobs))
+		for _, job := range jobs {
+			views = append(views, s.queueJob(job))
+		}
+		item.assign(views)
+	}
+	view.Poll = len(view.Running) > 0 || len(view.Queued) > 0
+	return view, nil
+}
+
+func (s *server) queueJob(job store.Job) queueJobView {
+	view := queueJobView{Job: job, CreatedAtText: job.CreatedAt.In(s.zone).Format("2006-01-02 15:04 MST")}
+	if job.Source == "subscription" {
+		view.SourceText = "自動"
+	} else {
+		view.SourceText = "手動"
+	}
+	if job.Phase != nil {
+		switch *job.Phase {
+		case "video":
+			view.PhaseText = "映像"
+		case "audio":
+			view.PhaseText = "音声"
+		case "merging":
+			view.PhaseText = "結合"
+		case "moving":
+			view.PhaseText = "移動"
+		}
+	}
+	if job.DownloadedBytes != nil && job.TotalBytes != nil && *job.TotalBytes > 0 {
+		percent := 100 * float64(*job.DownloadedBytes) / float64(*job.TotalBytes)
+		percent = max(0, min(percent, 100))
+		if job.TotalIsEstimate {
+			view.PercentText = fmt.Sprintf("約%.0f%%", percent)
+		} else {
+			view.PercentText = fmt.Sprintf("%.0f%%", percent)
+		}
+	}
+	if job.SpeedBPS != nil && *job.SpeedBPS > 0 {
+		view.SpeedText = formatBytes(*job.SpeedBPS) + "/秒"
+	}
+	if job.ETASeconds != nil && *job.ETASeconds >= 0 {
+		seconds := int64(*job.ETASeconds)
+		view.ETAText = fmt.Sprintf("残り%d分%d秒", seconds/60, seconds%60)
+	}
+	if job.FragmentIndex != nil && job.FragmentCount != nil && *job.FragmentCount > 0 {
+		view.FragmentText = fmt.Sprintf("断片 %d/%d", *job.FragmentIndex, *job.FragmentCount)
+	}
+	if job.ErrorKind != nil {
+		view.ErrorKindText = errorKindText(*job.ErrorKind)
+	}
+	return view
+}
+
+func formatBytes(n float64) string {
+	for _, unit := range []string{"B", "KiB", "MiB", "GiB"} {
+		if n < 1024 || unit == "GiB" {
+			return fmt.Sprintf("%.1f %s", n, unit)
+		}
+		n /= 1024
+	}
+	return ""
+}
+
+func errorKindText(kind string) string {
+	switch kind {
+	case "drm":
+		return "DRM のため取得不可"
+	case "geo":
+		return "地域制限"
+	case "network":
+		return "ネットワーク"
+	case "unavailable":
+		return "配信終了・存在しない"
+	case "extract":
+		return "抽出失敗"
+	case "invalid_output":
+		return "出力形式の不正"
+	default:
+		return "不明"
+	}
+}
+
+func (s *server) retryJob(w http.ResponseWriter, r *http.Request) {
+	s.queueMutation(w, r, s.jobs.Retry, false)
+}
+
+func (s *server) cancelJob(w http.ResponseWriter, r *http.Request) {
+	s.queueMutation(w, r, s.jobs.CancelQueued, true)
+}
+
+func (s *server) queueMutation(w http.ResponseWriter, r *http.Request, change func(context.Context, int64) error, queuedOnly bool) {
+	id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
+	if err != nil || id <= 0 {
+		http.Error(w, "ジョブ ID が不正です", http.StatusBadRequest)
+		return
+	}
+	if queuedOnly {
+		job, err := s.jobs.GetJob(r.Context(), id)
+		if errors.Is(err, pgx.ErrNoRows) {
+			http.Error(w, "ジョブが見つかりません", http.StatusNotFound)
+			return
+		}
+		if err != nil {
+			s.logger.Printf("ジョブ状態の取得に失敗: %v", err)
+			http.Error(w, "操作できませんでした", http.StatusInternalServerError)
+			return
+		}
+		if job.Status != "queued" {
+			http.Error(w, "待機中のジョブだけ取り消せます", http.StatusConflict)
+			return
+		}
+	}
+	if err := change(r.Context(), id); err != nil {
+		if errors.Is(err, store.ErrInvalidTransition) {
+			http.Error(w, "操作できる状態ではありません", http.StatusConflict)
+		} else {
+			s.logger.Printf("待ち行列の操作に失敗: %v", err)
+			http.Error(w, "操作できませんでした", http.StatusInternalServerError)
+		}
+		return
+	}
+	if r.Header.Get("HX-Request") != "true" {
+		http.Redirect(w, r, "/queue", http.StatusSeeOther)
+		return
+	}
+	s.queueFragment(w, r)
+}