fix: resolve 8 architecture flaws across download, playlist, and webhook
All checks were successful
CI / build (push) Successful in 52s

- Add --continue and --max-filesize to yt-dlp for download resilience
- Route thumbnail downloads through HTTP_PROXY
- Show per-video progress (i/N) with cancel for playlist downloads
- Run playlist asynchronously so cancel callbacks can be processed
- Guard handlePlaylistConfirm against concurrent active jobs
- Close Done channel in playlist entry lifecycle
- Add stepState TTL (30 min) to prevent stale states
- Wrap webhook server for graceful shutdown on SIGTERM
- Refactor formatBytes to loop-based implementation
- Show first line of stderr in user-facing error messages
- Fix format pointer reuse in playlist loop
- Add CreatedAt field to stepState for TTL tracking
- Add MaxFileSize and ProgressPrefix fields to DownloadJob
This commit is contained in:
2026-06-26 01:04:34 +03:30
parent 57dffc4194
commit 609237f4e3
7 changed files with 181 additions and 69 deletions

View File

@@ -5,6 +5,7 @@ import (
"log/slog" "log/slog"
"net/http" "net/http"
"os" "os"
"time"
tgbot "github.com/go-telegram/bot" tgbot "github.com/go-telegram/bot"
) )
@@ -23,10 +24,26 @@ func startWebhook(ctx context.Context, b *tgbot.Bot, webhookURL string) {
tlsCert := os.Getenv("BOT_TLS_CERT") tlsCert := os.Getenv("BOT_TLS_CERT")
tlsKey := os.Getenv("BOT_TLS_KEY") tlsKey := os.Getenv("BOT_TLS_KEY")
server := &http.Server{
Addr: listenAddr,
Handler: b.WebhookHandler(),
}
// Shutdown HTTP server gracefully when context is cancelled
go func() {
<-ctx.Done()
slog.Info("shutting down webhook server")
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := server.Shutdown(shutdownCtx); err != nil {
slog.Error("webhook server shutdown", "error", err)
}
}()
if tlsCert != "" && tlsKey != "" { if tlsCert != "" && tlsKey != "" {
slog.Info("starting HTTPS webhook", "addr", listenAddr) slog.Info("starting HTTPS webhook", "addr", listenAddr)
go func() { go func() {
if err := http.ListenAndServeTLS(listenAddr, tlsCert, tlsKey, b.WebhookHandler()); err != nil { if err := server.ListenAndServeTLS(tlsCert, tlsKey); err != nil && err != http.ErrServerClosed {
slog.Error("HTTPS server", "error", err) slog.Error("HTTPS server", "error", err)
os.Exit(1) os.Exit(1)
} }
@@ -34,7 +51,7 @@ func startWebhook(ctx context.Context, b *tgbot.Bot, webhookURL string) {
} else { } else {
slog.Info("starting HTTP webhook (TLS by reverse proxy)", "addr", listenAddr) slog.Info("starting HTTP webhook (TLS by reverse proxy)", "addr", listenAddr)
go func() { go func() {
if err := http.ListenAndServe(listenAddr, b.WebhookHandler()); err != nil { if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
slog.Error("HTTP server", "error", err) slog.Error("HTTP server", "error", err)
os.Exit(1) os.Exit(1)
} }

View File

@@ -10,6 +10,7 @@ import (
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
"strconv"
"strings" "strings"
"sync" "sync"
"syscall" "syscall"
@@ -305,6 +306,14 @@ func (c *Client) StartDownload(job *domain.DownloadJob, downloadDir string) (<-c
// --newline uses \n instead of \r so bufio.Scanner can read line-by-line // --newline uses \n instead of \r so bufio.Scanner can read line-by-line
args = append(args, "--progress", "--newline") args = append(args, "--progress", "--newline")
// Allow resuming partial downloads
args = append(args, "--continue")
// Enforce max file size limit
if job.MaxFileSize > 0 {
args = append(args, "--max-filesize", strconv.FormatInt(job.MaxFileSize, 10))
}
// Language selection if specified // Language selection if specified
if job.Language != "" { if job.Language != "" {
args = append(args, "--sub-langs", job.Language) args = append(args, "--sub-langs", job.Language)

View File

@@ -124,8 +124,10 @@ type DownloadJob struct {
Speed string Speed string
ETA string ETA string
FileSize int64 FileSize int64
MaxFileSize int64
FilePath string FilePath string
Title string Title string
ProgressPrefix string
QuotaApplied bool QuotaApplied bool
StderrLog string StderrLog string
Done chan struct{} Done chan struct{}

View File

@@ -31,12 +31,20 @@ type stepState struct {
FormatIDs []string FormatIDs []string
PlaylistState *domain.PlaylistState PlaylistState *domain.PlaylistState
SearchState *domain.SearchState SearchState *domain.SearchState
CreatedAt time.Time
} }
const stepStateTTL = 30 * time.Minute
func (h *Handler) getUserStepState(chatID int64) *stepState { func (h *Handler) getUserStepState(chatID int64) *stepState {
h.stepStatesMu.Lock() h.stepStatesMu.Lock()
defer h.stepStatesMu.Unlock() defer h.stepStatesMu.Unlock()
return h.stepStates[chatID] ss := h.stepStates[chatID]
if ss != nil && time.Since(ss.CreatedAt) > stepStateTTL {
delete(h.stepStates, chatID)
return nil
}
return ss
} }
func (h *Handler) setUserStepState(chatID int64, ss *stepState) { func (h *Handler) setUserStepState(chatID int64, ss *stepState) {
@@ -133,6 +141,7 @@ func (h *Handler) handleURLInput(ctx context.Context, chatID int64, text string,
URL: url, URL: url,
MediaInfo: info, MediaInfo: info,
Step: 0, Step: 0,
CreatedAt: time.Now(),
} }
h.sendMediaPreview(ctx, chatID, info) h.sendMediaPreview(ctx, chatID, info)
@@ -348,6 +357,7 @@ func (h *Handler) startDownload(ctx context.Context, chatID int64, msgID int, us
Status: domain.StatusDownloading, Status: domain.StatusDownloading,
Title: ss.MediaInfo.Title, Title: ss.MediaInfo.Title,
FileSize: fileSize, FileSize: fileSize,
MaxFileSize: h.maxFileSize,
Done: make(chan struct{}), Done: make(chan struct{}),
} }
@@ -375,9 +385,13 @@ func (h *Handler) runDownload(ctx context.Context, job *domain.DownloadJob) {
defer func() { <-h.downloadSem }() defer func() { <-h.downloadSem }()
kb := progressKeyboard(0, "", "") kb := progressKeyboard(0, "", "")
progressHeader := "Downloading: " + job.Title
if job.ProgressPrefix != "" {
progressHeader = job.ProgressPrefix + "\n" + progressHeader
}
msg, err := h.Bot.SendMessage(ctx, &tgbot.SendMessageParams{ msg, err := h.Bot.SendMessage(ctx, &tgbot.SendMessageParams{
ChatID: job.ChatID, ChatID: job.ChatID,
Text: fmt.Sprintf("Downloading: %s", job.Title), Text: progressHeader,
ReplyMarkup: &kb, ReplyMarkup: &kb,
}) })
if err != nil { if err != nil {
@@ -390,7 +404,9 @@ func (h *Handler) runDownload(ctx context.Context, job *domain.DownloadJob) {
if err != nil { if err != nil {
errMsg := err.Error() errMsg := err.Error()
if job.StderrLog != "" { if job.StderrLog != "" {
errMsg += ": " + job.StderrLog if lines := strings.SplitN(job.StderrLog, "\n", 2); len(lines) > 0 && lines[0] != "" {
errMsg = strings.TrimSpace(lines[0])
}
} }
h.editText(ctx, job.ChatID, job.MessageID, fmt.Sprintf("Download failed: %s", errMsg), nil) h.editText(ctx, job.ChatID, job.MessageID, fmt.Sprintf("Download failed: %s", errMsg), nil)
job.Status = domain.StatusFailed job.Status = domain.StatusFailed
@@ -460,7 +476,10 @@ func (h *Handler) trackDownloadProgress(ctx context.Context, job *domain.Downloa
} }
kb := progressKeyboard(upd.Progress, upd.Speed, upd.ETA) kb := progressKeyboard(upd.Progress, upd.Speed, upd.ETA)
text := fmt.Sprintf("Downloading: %s", job.Title) text := "Downloading: " + job.Title
if job.ProgressPrefix != "" {
text = job.ProgressPrefix + "\n" + text
}
h.editText(ctx, job.ChatID, job.MessageID, text, &kb) h.editText(ctx, job.ChatID, job.MessageID, text, &kb)
case <-ctx.Done(): case <-ctx.Done():
@@ -732,7 +751,13 @@ func (h *Handler) sendMediaPreview(ctx context.Context, chatID int64, info *doma
// downloadFile downloads a URL to a temp file in the given directory and returns the path. // downloadFile downloads a URL to a temp file in the given directory and returns the path.
func downloadFile(url, dir string) (string, error) { func downloadFile(url, dir string) (string, error) {
client := &http.Client{Timeout: 30 * time.Second} tr := &http.Transport{
Proxy: http.ProxyFromEnvironment,
}
client := &http.Client{
Timeout: 30 * time.Second,
Transport: tr,
}
resp, err := client.Get(url) resp, err := client.Get(url)
if err != nil { if err != nil {
return "", fmt.Errorf("http get: %w", err) return "", fmt.Errorf("http get: %w", err)

View File

@@ -464,21 +464,15 @@ func formatBytes(bytes int64) string {
if bytes < unit { if bytes < unit {
return strconv.FormatInt(bytes, 10) + "B" return strconv.FormatInt(bytes, 10) + "B"
} }
div, exp := int64(unit), 0 units := []string{"KB", "MB", "GB", "TB"}
for n := bytes / unit; n >= unit; n /= unit { size := float64(bytes)
div *= unit for _, u := range units {
exp++ size /= unit
if size < unit {
return fmt.Sprintf("%.1f%s", size, u)
} }
switch exp {
case 0:
return fmt.Sprintf("%.1fKB", float64(bytes)/float64(div))
case 1:
return fmt.Sprintf("%.1fMB", float64(bytes)/float64(div))
case 2:
return fmt.Sprintf("%.1fGB", float64(bytes)/float64(div))
default:
return fmt.Sprintf("%.1fTB", float64(bytes)/float64(div))
} }
return fmt.Sprintf("%.1fTB", size)
} }
func (h *Handler) generateJobID() string { func (h *Handler) generateJobID() string {

View File

@@ -3,7 +3,12 @@ package handler
import ( import (
"context" "context"
"fmt" "fmt"
"log/slog"
"strconv" "strconv"
"strings"
"time"
tgbot "github.com/go-telegram/bot"
"uptodownBot/internal/domain" "uptodownBot/internal/domain"
) )
@@ -20,6 +25,7 @@ func (h *Handler) handlePlaylist(ctx context.Context, chatID int64, userID int64
URL: url, URL: url,
MediaInfo: info, MediaInfo: info,
PlaylistState: state, PlaylistState: state,
CreatedAt: time.Now(),
} }
h.setUserStepState(chatID, ss) h.setUserStepState(chatID, ss)
@@ -110,6 +116,11 @@ func (h *Handler) handlePlaylistSelectAll(ctx context.Context, chatID int64, msg
} }
func (h *Handler) handlePlaylistConfirm(ctx context.Context, chatID int64, msgID int, userID int64) { func (h *Handler) handlePlaylistConfirm(ctx context.Context, chatID int64, msgID int, userID int64) {
if job := h.getActiveJob(userID); job != nil {
h.sendText(ctx, chatID, "You already have an active download. Cancel it first.")
return
}
state := h.getPlaylistState(chatID) state := h.getPlaylistState(chatID)
if state == nil { if state == nil {
return return
@@ -127,14 +138,24 @@ func (h *Handler) handlePlaylistConfirm(ctx context.Context, chatID int64, msgID
} }
h.clearStepState(chatID) h.clearStepState(chatID)
h.editText(ctx, chatID, msgID, fmt.Sprintf("Downloading %d videos sequentially...", len(selected)), nil)
// Parent context for the entire playlist — cancelling this stops all entries
playlistCtx, playlistCancel := context.WithCancel(context.Background())
h.setCancelFunc(userID, playlistCancel)
h.editText(ctx, chatID, msgID, fmt.Sprintf("Downloading %d videos...", len(selected)), nil)
// Run the playlist loop in a goroutine so the handler is free to process
// cancel callbacks and other updates concurrently.
go func() {
for i, entry := range selected { for i, entry := range selected {
h.sendText(ctx, chatID, fmt.Sprintf("Downloading (%d/%d): %s", i+1, len(selected), entry.Title)) if playlistCtx.Err() != nil {
break
}
info, err := h.YtDlp.GetMediaInfo(entry.URL) info, err := h.YtDlp.GetMediaInfo(entry.URL)
if err != nil { if err != nil {
h.sendText(ctx, chatID, fmt.Sprintf("Failed: %s", err.Error())) h.sendText(playlistCtx, chatID, fmt.Sprintf("Playlist entry (%d/%d) failed: %s", i+1, len(selected), err.Error()))
continue continue
} }
@@ -151,41 +172,84 @@ func (h *Handler) handlePlaylistConfirm(ctx context.Context, chatID int64, msgID
if prefs.DefaultMediaType == domain.MediaTypeAudio { if prefs.DefaultMediaType == domain.MediaTypeAudio {
formatID = prefs.DefaultAudioFormat formatID = prefs.DefaultAudioFormat
} }
format := &domain.Format{ID: formatID} ff := &domain.Format{ID: formatID}
for _, f := range info.Formats { for i := range info.Formats {
if f.ID == formatID { if info.Formats[i].ID == formatID {
format = &f ff = &info.Formats[i]
break break
} }
} }
dlCtx, dlCancel := context.WithCancel(context.Background())
job := &domain.DownloadJob{ job := &domain.DownloadJob{
ID: h.generateJobID(), ID: h.generateJobID(),
UserID: userID, UserID: userID,
ChatID: chatID, ChatID: chatID,
MessageID: 0,
URL: entry.URL, URL: entry.URL,
MediaType: prefs.DefaultMediaType, MediaType: prefs.DefaultMediaType,
SelectedFormat: format, SelectedFormat: ff,
FormatString: "",
Language: prefs.DefaultLanguage, Language: prefs.DefaultLanguage,
Status: domain.StatusDownloading, Status: domain.StatusDownloading,
Title: entry.Title, Title: entry.Title,
FileSize: format.Filesize, FileSize: ff.Filesize,
MaxFileSize: h.maxFileSize,
ProgressPrefix: fmt.Sprintf("Downloading (%d/%d)", i+1, len(selected)),
Done: make(chan struct{}), Done: make(chan struct{}),
} }
h.setActiveJob(userID, job) ok := h.downloadPlaylistEntry(playlistCtx, userID, chatID, job)
h.setCancelFunc(userID, dlCancel) if !ok {
break
h.runDownload(dlCtx, job) }
// runDownload is synchronous — it blocks until the download completes.
// The Done channel is closed inside runDownload when it exits.
// No busy-polling needed.
} }
h.sendText(ctx, chatID, "All playlist downloads completed.") h.clearActiveJob(userID)
h.sendText(playlistCtx, chatID, "Playlist downloads finished.")
}()
}
// downloadPlaylistEntry handles a single entry in a playlist download.
// It returns false if the playlist should stop (cancelled/fatal).
func (h *Handler) downloadPlaylistEntry(ctx context.Context, userID, chatID int64, job *domain.DownloadJob) bool {
if job.Done != nil {
defer close(job.Done)
}
h.setActiveJob(userID, job)
kb := progressKeyboard(0, "", "")
progressHeader := job.ProgressPrefix + "\nDownloading: " + job.Title
msg, sendErr := h.Bot.SendMessage(ctx, &tgbot.SendMessageParams{
ChatID: chatID,
Text: progressHeader,
ReplyMarkup: &kb,
})
if sendErr != nil {
slog.Error("send playlist progress msg failed", "error", sendErr)
return true // non-fatal, continue to next
}
job.MessageID = msg.ID
updates, dlErr := h.YtDlp.StartDownload(job, h.downloadDir)
if dlErr != nil {
errMsg := dlErr.Error()
if job.StderrLog != "" {
if lines := strings.SplitN(job.StderrLog, "\n", 2); len(lines) > 0 && lines[0] != "" {
errMsg = strings.TrimSpace(lines[0])
}
}
h.editText(ctx, chatID, job.MessageID, fmt.Sprintf("Download failed: %s", errMsg), nil)
return true
}
h.trackDownloadProgress(ctx, job, updates)
if job.Status == domain.StatusCompleted {
h.handleDownloadCompletion(ctx, job)
}
return job.Status != domain.StatusCancelled
} }
func countSelected(state *domain.PlaylistState) int { func countSelected(state *domain.PlaylistState) int {

View File

@@ -5,6 +5,7 @@ import (
"fmt" "fmt"
"strconv" "strconv"
"strings" "strings"
"time"
"github.com/go-telegram/bot/models" "github.com/go-telegram/bot/models"
@@ -39,7 +40,7 @@ func (h *Handler) handleSearchQuery(ctx context.Context, chatID int64, text stri
PerPage: 5, PerPage: 5,
Source: source, Source: source,
} }
ss := &stepState{SearchState: state} ss := &stepState{SearchState: state, CreatedAt: time.Now()}
h.setUserStepState(chatID, ss) h.setUserStepState(chatID, ss)
kb := searchResultsKeyboard(state) kb := searchResultsKeyboard(state)