feat: torrent streaming with hls transcoding (#1)
* feat: add ffmpeg for hls streaming * feat: torrent streaming with hls transcoding - add nyaa.si torrent search client - add streaming service using anacrolix/torrent - add hls transcoding via ffmpeg for browser playback - add watch page with episode selection - add socks5 proxy support via TORRENT_PROXY env - switch to modernc.org/sqlite (pure go, no cgo conflicts) - update dockerfile with ffmpeg
This commit is contained in:
GitHub
349
internal/streaming/handler.go
Normal file
349
internal/streaming/handler.go
Normal file
@@ -0,0 +1,349 @@
|
||||
package streaming
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"html"
|
||||
"log"
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"mal/internal/jikan"
|
||||
"mal/internal/nyaa"
|
||||
"mal/internal/templates"
|
||||
)
|
||||
|
||||
type Handler struct {
|
||||
svc *Service
|
||||
jikan *jikan.Client
|
||||
}
|
||||
|
||||
func NewHandler(svc *Service, jikanClient *jikan.Client) *Handler {
|
||||
return &Handler{svc: svc, jikan: jikanClient}
|
||||
}
|
||||
|
||||
func (h *Handler) RegisterRoutes(mux *http.ServeMux) {
|
||||
// Watch page
|
||||
mux.HandleFunc("GET /watch/{animeID}/{episode}", h.HandleWatchPage)
|
||||
|
||||
// Search endpoints
|
||||
mux.HandleFunc("GET /api/stream/search", h.HandleSearch)
|
||||
mux.HandleFunc("GET /api/stream/search/episode", h.HandleSearchEpisode)
|
||||
mux.HandleFunc("GET /api/stream/search-htmx", h.HandleSearchHTMX)
|
||||
|
||||
// Streaming endpoints
|
||||
mux.HandleFunc("POST /api/stream/start", h.HandleStartStream)
|
||||
mux.HandleFunc("GET /api/stream/video/{hash}", h.HandleStreamVideo)
|
||||
mux.HandleFunc("GET /api/stream/info/{hash}", h.HandleStreamInfo)
|
||||
mux.HandleFunc("DELETE /api/stream/{hash}", h.HandleDropStream)
|
||||
|
||||
// HLS endpoints
|
||||
mux.HandleFunc("POST /api/stream/hls/{hash}", h.HandleStartHLS)
|
||||
mux.HandleFunc("GET /api/stream/hls/{hash}/playlist.m3u8", h.HandleHLSPlaylist)
|
||||
mux.HandleFunc("GET /api/stream/hls/{hash}/{segment}", h.HandleHLSSegment)
|
||||
}
|
||||
|
||||
// HandleWatchPage renders the video player page for an episode
|
||||
func (h *Handler) HandleWatchPage(w http.ResponseWriter, r *http.Request) {
|
||||
animeIDStr := r.PathValue("animeID")
|
||||
episodeStr := r.PathValue("episode")
|
||||
|
||||
animeID, err := strconv.Atoi(animeIDStr)
|
||||
if err != nil || animeID <= 0 {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
episode, err := strconv.Atoi(episodeStr)
|
||||
if err != nil || episode <= 0 {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
anime, err := h.jikan.GetAnimeByID(animeID)
|
||||
if err != nil {
|
||||
log.Printf("failed to get anime %d: %v", animeID, err)
|
||||
http.Error(w, "Failed to fetch anime", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
// Search for torrents for this episode
|
||||
torrents, err := h.svc.SearchEpisode(anime.Title, episode)
|
||||
if err != nil {
|
||||
log.Printf("torrent search error: %v", err)
|
||||
torrents = nil
|
||||
}
|
||||
|
||||
// Filter to 1080p by default, fallback to all
|
||||
filtered := nyaa.FilterByQuality(torrents, "1080p")
|
||||
if len(filtered) == 0 {
|
||||
filtered = torrents
|
||||
}
|
||||
|
||||
// Limit to top 10
|
||||
if len(filtered) > 10 {
|
||||
filtered = filtered[:10]
|
||||
}
|
||||
|
||||
templates.WatchPage(anime, episode, filtered).Render(r.Context(), w)
|
||||
}
|
||||
|
||||
// HandleSearch searches nyaa for anime torrents
|
||||
func (h *Handler) HandleSearch(w http.ResponseWriter, r *http.Request) {
|
||||
query := r.URL.Query().Get("q")
|
||||
if query == "" {
|
||||
http.Error(w, "query required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
torrents, err := h.svc.SearchAnime(query)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
quality := r.URL.Query().Get("quality")
|
||||
if quality != "" {
|
||||
torrents = nyaa.FilterByQuality(torrents, quality)
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(torrents)
|
||||
}
|
||||
|
||||
// HandleSearchEpisode searches nyaa for a specific episode
|
||||
func (h *Handler) HandleSearchEpisode(w http.ResponseWriter, r *http.Request) {
|
||||
title := r.URL.Query().Get("title")
|
||||
episodeStr := r.URL.Query().Get("episode")
|
||||
|
||||
if title == "" || episodeStr == "" {
|
||||
http.Error(w, "title and episode required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
episode, err := strconv.Atoi(episodeStr)
|
||||
if err != nil {
|
||||
http.Error(w, "invalid episode number", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
torrents, err := h.svc.SearchEpisode(title, episode)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
quality := r.URL.Query().Get("quality")
|
||||
if quality != "" {
|
||||
torrents = nyaa.FilterByQuality(torrents, quality)
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(torrents)
|
||||
}
|
||||
|
||||
type startStreamRequest struct {
|
||||
Magnet string `json:"magnet"`
|
||||
}
|
||||
|
||||
type startStreamResponse struct {
|
||||
InfoHash string `json:"info_hash"`
|
||||
Name string `json:"name"`
|
||||
Size int64 `json:"size"`
|
||||
}
|
||||
|
||||
// HandleStartStream starts streaming a torrent from a magnet link
|
||||
func (h *Handler) HandleStartStream(w http.ResponseWriter, r *http.Request) {
|
||||
var req startStreamRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
if req.Magnet == "" {
|
||||
http.Error(w, "magnet required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
info, err := h.svc.AddMagnet(r.Context(), req.Magnet)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(startStreamResponse{
|
||||
InfoHash: info.InfoHash,
|
||||
Name: info.Name,
|
||||
Size: info.Size,
|
||||
})
|
||||
}
|
||||
|
||||
// HandleStreamVideo streams the video file from a torrent
|
||||
func (h *Handler) HandleStreamVideo(w http.ResponseWriter, r *http.Request) {
|
||||
hash := r.PathValue("hash")
|
||||
if hash == "" {
|
||||
http.Error(w, "hash required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.svc.StreamVideo(w, r, hash); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// HandleStreamInfo returns info about an active stream
|
||||
func (h *Handler) HandleStreamInfo(w http.ResponseWriter, r *http.Request) {
|
||||
hash := r.PathValue("hash")
|
||||
if hash == "" {
|
||||
http.Error(w, "hash required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
info, err := h.svc.GetStreamInfo(hash)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(info)
|
||||
}
|
||||
|
||||
// HandleDropStream stops and removes a stream
|
||||
func (h *Handler) HandleDropStream(w http.ResponseWriter, r *http.Request) {
|
||||
hash := r.PathValue("hash")
|
||||
if hash == "" {
|
||||
http.Error(w, "hash required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
h.svc.DropTorrent(hash)
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
// TorrentResult for HTMX responses
|
||||
type TorrentResult struct {
|
||||
Torrents []nyaa.Torrent
|
||||
Query string
|
||||
Episode int
|
||||
}
|
||||
|
||||
// HandleSearchHTMX returns HTML for torrent search results
|
||||
func (h *Handler) HandleSearchHTMX(w http.ResponseWriter, r *http.Request) {
|
||||
query := r.URL.Query().Get("q")
|
||||
if query == "" {
|
||||
fmt.Fprint(w, `<div class="search-empty">enter a search query</div>`)
|
||||
return
|
||||
}
|
||||
|
||||
torrents, err := h.svc.SearchAnime(query)
|
||||
if err != nil {
|
||||
fmt.Fprintf(w, `<div class="search-error">error: %s</div>`, html.EscapeString(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
quality := r.URL.Query().Get("quality")
|
||||
if quality != "" {
|
||||
torrents = nyaa.FilterByQuality(torrents, quality)
|
||||
}
|
||||
|
||||
// Filter to only torrents with magnet links
|
||||
var withMagnets []nyaa.Torrent
|
||||
for _, t := range torrents {
|
||||
if t.Magnet != "" {
|
||||
withMagnets = append(withMagnets, t)
|
||||
}
|
||||
}
|
||||
|
||||
if len(withMagnets) == 0 {
|
||||
fmt.Fprint(w, `<div class="search-empty">no torrents found (or no magnet links available)</div>`)
|
||||
return
|
||||
}
|
||||
|
||||
// Return simple HTML list
|
||||
fmt.Fprint(w, `<div class="torrent-list">`)
|
||||
for _, t := range withMagnets {
|
||||
fmt.Fprintf(w, `
|
||||
<div class="torrent-item" data-magnet="%s">
|
||||
<div class="torrent-title">%s</div>
|
||||
<div class="torrent-meta">
|
||||
<span class="torrent-size">%s</span>
|
||||
<span class="torrent-seeds">%d seeds</span>
|
||||
</div>
|
||||
</div>
|
||||
`, html.EscapeString(t.Magnet), html.EscapeString(t.Title), html.EscapeString(t.Size), t.Seeders)
|
||||
}
|
||||
fmt.Fprint(w, `</div>`)
|
||||
}
|
||||
|
||||
// HandleStartHLS starts HLS transcoding for a torrent
|
||||
func (h *Handler) HandleStartHLS(w http.ResponseWriter, r *http.Request) {
|
||||
hash := r.PathValue("hash")
|
||||
if hash == "" {
|
||||
http.Error(w, "hash required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Check torrent exists
|
||||
if _, ok := h.svc.GetTorrent(hash); !ok {
|
||||
http.Error(w, "torrent not found - start stream first", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
session, err := h.svc.StartHLS(r.Context(), hash)
|
||||
if err != nil {
|
||||
log.Printf("HLS start error: %v", err)
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]string{
|
||||
"playlist": fmt.Sprintf("/api/stream/hls/%s/playlist.m3u8", hash),
|
||||
"status": "ready",
|
||||
"output": session.OutputDir,
|
||||
})
|
||||
}
|
||||
|
||||
// HandleHLSPlaylist serves the HLS playlist file
|
||||
func (h *Handler) HandleHLSPlaylist(w http.ResponseWriter, r *http.Request) {
|
||||
hash := r.PathValue("hash")
|
||||
if hash == "" {
|
||||
http.Error(w, "hash required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
session, ok := h.svc.GetHLSSession(hash)
|
||||
if !ok {
|
||||
http.Error(w, "HLS session not found - start transcoding first", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
http.ServeFile(w, r, session.Playlist)
|
||||
}
|
||||
|
||||
// HandleHLSSegment serves an HLS segment file
|
||||
func (h *Handler) HandleHLSSegment(w http.ResponseWriter, r *http.Request) {
|
||||
hash := r.PathValue("hash")
|
||||
segment := r.PathValue("segment")
|
||||
|
||||
if hash == "" || segment == "" {
|
||||
http.Error(w, "hash and segment required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
session, ok := h.svc.GetHLSSession(hash)
|
||||
if !ok {
|
||||
http.Error(w, "HLS session not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
// Serve the segment file
|
||||
segmentPath := fmt.Sprintf("%s/%s", session.OutputDir, segment)
|
||||
w.Header().Set("Content-Type", "video/mp2t")
|
||||
http.ServeFile(w, r, segmentPath)
|
||||
}
|
||||
248
internal/streaming/hls.go
Normal file
248
internal/streaming/hls.go
Normal file
@@ -0,0 +1,248 @@
|
||||
package streaming
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// HLSTranscoder manages on-demand HLS transcoding sessions
|
||||
type HLSTranscoder struct {
|
||||
logger *slog.Logger
|
||||
sessions map[string]*HLSSession
|
||||
mu sync.RWMutex
|
||||
baseDir string
|
||||
}
|
||||
|
||||
// HLSSession represents an active transcoding session
|
||||
type HLSSession struct {
|
||||
InfoHash string
|
||||
OutputDir string
|
||||
Playlist string
|
||||
cmd *exec.Cmd
|
||||
cancel context.CancelFunc
|
||||
ready chan struct{}
|
||||
err error
|
||||
lastAccess time.Time
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// NewHLSTranscoder creates a new HLS transcoder
|
||||
func NewHLSTranscoder(logger *slog.Logger) (*HLSTranscoder, error) {
|
||||
// Check ffmpeg is available
|
||||
if _, err := exec.LookPath("ffmpeg"); err != nil {
|
||||
return nil, fmt.Errorf("ffmpeg not found in PATH: %w", err)
|
||||
}
|
||||
|
||||
baseDir := filepath.Join(os.TempDir(), "mal-hls")
|
||||
if err := os.MkdirAll(baseDir, 0755); err != nil {
|
||||
return nil, fmt.Errorf("failed to create HLS temp dir: %w", err)
|
||||
}
|
||||
|
||||
t := &HLSTranscoder{
|
||||
logger: logger,
|
||||
sessions: make(map[string]*HLSSession),
|
||||
baseDir: baseDir,
|
||||
}
|
||||
|
||||
// Start cleanup goroutine
|
||||
go t.cleanupLoop()
|
||||
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// StartSessionWithReader starts transcoding from a reader (piped input) to HLS
|
||||
func (t *HLSTranscoder) StartSessionWithReader(infoHash string, reader io.Reader) (*HLSSession, error) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
|
||||
// Return existing session if available
|
||||
if session, ok := t.sessions[infoHash]; ok {
|
||||
session.mu.Lock()
|
||||
session.lastAccess = time.Now()
|
||||
session.mu.Unlock()
|
||||
return session, nil
|
||||
}
|
||||
|
||||
// Create output directory
|
||||
outputDir := filepath.Join(t.baseDir, infoHash)
|
||||
if err := os.MkdirAll(outputDir, 0755); err != nil {
|
||||
return nil, fmt.Errorf("failed to create output dir: %w", err)
|
||||
}
|
||||
|
||||
playlist := filepath.Join(outputDir, "stream.m3u8")
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
session := &HLSSession{
|
||||
InfoHash: infoHash,
|
||||
OutputDir: outputDir,
|
||||
Playlist: playlist,
|
||||
cancel: cancel,
|
||||
ready: make(chan struct{}),
|
||||
lastAccess: time.Now(),
|
||||
}
|
||||
|
||||
// Start ffmpeg with pipe input
|
||||
// Use browser-compatible H.264 baseline/main profile and AAC-LC
|
||||
cmd := exec.CommandContext(ctx, "ffmpeg",
|
||||
"-i", "pipe:0", // Read from stdin
|
||||
// Video: H.264 with browser-compatible settings
|
||||
"-c:v", "libx264",
|
||||
"-preset", "veryfast",
|
||||
"-crf", "23",
|
||||
"-profile:v", "main",
|
||||
"-level", "4.0",
|
||||
"-pix_fmt", "yuv420p",
|
||||
// Audio: AAC-LC stereo
|
||||
"-c:a", "aac",
|
||||
"-b:a", "128k",
|
||||
"-ac", "2",
|
||||
"-ar", "44100",
|
||||
// HLS output
|
||||
"-f", "hls",
|
||||
"-hls_time", "4",
|
||||
"-hls_list_size", "0",
|
||||
"-hls_flags", "append_list+independent_segments",
|
||||
"-hls_segment_type", "mpegts",
|
||||
"-hls_segment_filename", filepath.Join(outputDir, "segment_%03d.ts"),
|
||||
"-start_number", "0",
|
||||
playlist,
|
||||
)
|
||||
|
||||
// Pipe the reader to ffmpeg's stdin
|
||||
cmd.Stdin = reader
|
||||
session.cmd = cmd
|
||||
|
||||
// Capture stderr for debugging
|
||||
cmd.Stderr = &ffmpegLogger{logger: t.logger, infoHash: infoHash}
|
||||
|
||||
if err := cmd.Start(); err != nil {
|
||||
cancel()
|
||||
os.RemoveAll(outputDir)
|
||||
return nil, fmt.Errorf("failed to start ffmpeg: %w", err)
|
||||
}
|
||||
|
||||
t.sessions[infoHash] = session
|
||||
|
||||
// Wait for playlist to be created
|
||||
go func() {
|
||||
defer close(session.ready)
|
||||
|
||||
for i := 0; i < 120; i++ { // Wait up to 60 seconds
|
||||
if _, err := os.Stat(playlist); err == nil {
|
||||
// Check if at least one segment exists
|
||||
segments, _ := filepath.Glob(filepath.Join(outputDir, "segment_*.ts"))
|
||||
if len(segments) > 0 {
|
||||
t.logger.Info("HLS segments ready", "hash", infoHash, "segments", len(segments))
|
||||
return
|
||||
}
|
||||
}
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
}
|
||||
|
||||
session.err = fmt.Errorf("timeout waiting for HLS segments")
|
||||
cancel()
|
||||
}()
|
||||
|
||||
// Monitor process
|
||||
go func() {
|
||||
err := cmd.Wait()
|
||||
if err != nil && ctx.Err() == nil {
|
||||
t.logger.Error("ffmpeg exited with error", "hash", infoHash, "error", err)
|
||||
} else {
|
||||
t.logger.Info("ffmpeg finished", "hash", infoHash)
|
||||
}
|
||||
}()
|
||||
|
||||
return session, nil
|
||||
}
|
||||
|
||||
// GetSession returns an existing session
|
||||
func (t *HLSTranscoder) GetSession(infoHash string) (*HLSSession, bool) {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
|
||||
session, ok := t.sessions[infoHash]
|
||||
if ok {
|
||||
session.mu.Lock()
|
||||
session.lastAccess = time.Now()
|
||||
session.mu.Unlock()
|
||||
}
|
||||
return session, ok
|
||||
}
|
||||
|
||||
// StopSession stops a transcoding session
|
||||
func (t *HLSTranscoder) StopSession(infoHash string) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
|
||||
if session, ok := t.sessions[infoHash]; ok {
|
||||
session.cancel()
|
||||
os.RemoveAll(session.OutputDir)
|
||||
delete(t.sessions, infoHash)
|
||||
}
|
||||
}
|
||||
|
||||
// WaitReady waits for the session to have segments ready
|
||||
func (s *HLSSession) WaitReady(ctx context.Context) error {
|
||||
select {
|
||||
case <-s.ready:
|
||||
return s.err
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// cleanupLoop removes stale sessions
|
||||
func (t *HLSTranscoder) cleanupLoop() {
|
||||
ticker := time.NewTicker(5 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
|
||||
for range ticker.C {
|
||||
t.mu.Lock()
|
||||
now := time.Now()
|
||||
for hash, session := range t.sessions {
|
||||
session.mu.Lock()
|
||||
if now.Sub(session.lastAccess) > 10*time.Minute {
|
||||
session.cancel()
|
||||
os.RemoveAll(session.OutputDir)
|
||||
delete(t.sessions, hash)
|
||||
t.logger.Info("cleaned up stale HLS session", "hash", hash)
|
||||
}
|
||||
session.mu.Unlock()
|
||||
}
|
||||
t.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// Shutdown stops all sessions
|
||||
func (t *HLSTranscoder) Shutdown() {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
|
||||
for hash, session := range t.sessions {
|
||||
session.cancel()
|
||||
os.RemoveAll(session.OutputDir)
|
||||
delete(t.sessions, hash)
|
||||
}
|
||||
|
||||
os.RemoveAll(t.baseDir)
|
||||
}
|
||||
|
||||
// ffmpegLogger logs ffmpeg stderr output
|
||||
type ffmpegLogger struct {
|
||||
logger *slog.Logger
|
||||
infoHash string
|
||||
}
|
||||
|
||||
func (l *ffmpegLogger) Write(p []byte) (n int, err error) {
|
||||
l.logger.Debug("ffmpeg", "hash", l.infoHash, "output", string(p))
|
||||
return len(p), nil
|
||||
}
|
||||
423
internal/streaming/service.go
Normal file
423
internal/streaming/service.go
Normal file
@@ -0,0 +1,423 @@
|
||||
package streaming
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/anacrolix/torrent"
|
||||
"github.com/anacrolix/torrent/metainfo"
|
||||
"golang.org/x/net/proxy"
|
||||
|
||||
"mal/internal/nyaa"
|
||||
)
|
||||
|
||||
type Service struct {
|
||||
client *torrent.Client
|
||||
nyaa *nyaa.Client
|
||||
hls *HLSTranscoder
|
||||
mu sync.RWMutex
|
||||
activeTorrents map[string]*torrent.Torrent
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
type StreamInfo struct {
|
||||
InfoHash string
|
||||
Name string
|
||||
Size int64
|
||||
Files []FileInfo
|
||||
Progress float64
|
||||
DownloadRate int64
|
||||
Peers int
|
||||
}
|
||||
|
||||
type FileInfo struct {
|
||||
Index int
|
||||
Path string
|
||||
Size int64
|
||||
}
|
||||
|
||||
func NewService(logger *slog.Logger) (*Service, error) {
|
||||
cfg := torrent.NewDefaultClientConfig()
|
||||
cfg.ListenPort = 42069
|
||||
cfg.Seed = false
|
||||
cfg.NoUpload = true
|
||||
|
||||
// Use temp directory for downloads
|
||||
cfg.DataDir = filepath.Join("/tmp", "mal-streams")
|
||||
|
||||
// Configure SOCKS5 proxy if TORRENT_PROXY is set
|
||||
// Usage: export TORRENT_PROXY=socks5://127.0.0.1:1080
|
||||
// Start with: ssh -D 1080 -N user@your-vps
|
||||
if proxyURL := os.Getenv("TORRENT_PROXY"); proxyURL != "" {
|
||||
parsed, err := url.Parse(proxyURL)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid TORRENT_PROXY url: %w", err)
|
||||
}
|
||||
|
||||
dialer, err := proxy.FromURL(parsed, proxy.Direct)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create proxy dialer: %w", err)
|
||||
}
|
||||
|
||||
// Proxy HTTP requests (trackers, webseeds)
|
||||
cfg.HTTPProxy = func(*http.Request) (*url.URL, error) {
|
||||
return parsed, nil
|
||||
}
|
||||
|
||||
// Proxy peer connections via DialContext
|
||||
if contextDialer, ok := dialer.(proxy.ContextDialer); ok {
|
||||
cfg.HTTPDialContext = contextDialer.DialContext
|
||||
}
|
||||
|
||||
logger.Info("torrent proxy configured", "proxy", proxyURL)
|
||||
}
|
||||
|
||||
client, err := torrent.NewClient(cfg)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create torrent client: %w", err)
|
||||
}
|
||||
|
||||
hls, err := NewHLSTranscoder(logger)
|
||||
if err != nil {
|
||||
logger.Warn("HLS transcoding unavailable", "error", err)
|
||||
// Continue without HLS - will fall back to direct streaming
|
||||
}
|
||||
|
||||
return &Service{
|
||||
client: client,
|
||||
nyaa: nyaa.NewClient(),
|
||||
hls: hls,
|
||||
activeTorrents: make(map[string]*torrent.Torrent),
|
||||
logger: logger,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// SearchEpisode searches nyaa for torrents of a specific episode
|
||||
func (s *Service) SearchEpisode(animeTitle string, episode int) ([]nyaa.Torrent, error) {
|
||||
return s.nyaa.SearchEpisode(animeTitle, episode)
|
||||
}
|
||||
|
||||
// SearchAnime searches nyaa for torrents of an anime
|
||||
func (s *Service) SearchAnime(query string) ([]nyaa.Torrent, error) {
|
||||
return s.nyaa.SearchAnime(query)
|
||||
}
|
||||
|
||||
// AddMagnet adds a magnet link and returns stream info
|
||||
func (s *Service) AddMagnet(ctx context.Context, magnetURI string) (*StreamInfo, error) {
|
||||
t, err := s.client.AddMagnet(magnetURI)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to add magnet: %w", err)
|
||||
}
|
||||
|
||||
// Wait for metadata with timeout
|
||||
select {
|
||||
case <-t.GotInfo():
|
||||
case <-ctx.Done():
|
||||
t.Drop()
|
||||
return nil, ctx.Err()
|
||||
case <-time.After(60 * time.Second):
|
||||
t.Drop()
|
||||
return nil, fmt.Errorf("timeout waiting for torrent metadata")
|
||||
}
|
||||
|
||||
infoHash := t.InfoHash().HexString()
|
||||
|
||||
s.mu.Lock()
|
||||
s.activeTorrents[infoHash] = t
|
||||
s.mu.Unlock()
|
||||
|
||||
return s.getStreamInfo(t), nil
|
||||
}
|
||||
|
||||
// GetTorrent returns an active torrent by info hash
|
||||
func (s *Service) GetTorrent(infoHash string) (*torrent.Torrent, bool) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
t, ok := s.activeTorrents[infoHash]
|
||||
return t, ok
|
||||
}
|
||||
|
||||
// StreamFile streams a specific file from a torrent
|
||||
func (s *Service) StreamFile(w http.ResponseWriter, r *http.Request, infoHash string, fileIdx int) error {
|
||||
t, ok := s.GetTorrent(infoHash)
|
||||
if !ok {
|
||||
return fmt.Errorf("torrent not found: %s", infoHash)
|
||||
}
|
||||
|
||||
files := t.Files()
|
||||
if fileIdx < 0 || fileIdx >= len(files) {
|
||||
return fmt.Errorf("invalid file index: %d", fileIdx)
|
||||
}
|
||||
|
||||
file := files[fileIdx]
|
||||
reader := file.NewReader()
|
||||
reader.SetReadahead(file.Length() / 100) // 1% readahead
|
||||
reader.SetResponsive()
|
||||
|
||||
// Determine content type
|
||||
contentType := "application/octet-stream"
|
||||
ext := strings.ToLower(filepath.Ext(file.Path()))
|
||||
switch ext {
|
||||
case ".mp4":
|
||||
contentType = "video/mp4"
|
||||
case ".mkv":
|
||||
contentType = "video/x-matroska"
|
||||
case ".webm":
|
||||
contentType = "video/webm"
|
||||
case ".avi":
|
||||
contentType = "video/x-msvideo"
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", contentType)
|
||||
w.Header().Set("Accept-Ranges", "bytes")
|
||||
|
||||
// Handle range requests for seeking
|
||||
http.ServeContent(w, r, file.Path(), time.Time{}, reader)
|
||||
return nil
|
||||
}
|
||||
|
||||
// StreamVideo finds and streams the main video file from a torrent
|
||||
func (s *Service) StreamVideo(w http.ResponseWriter, r *http.Request, infoHash string) error {
|
||||
t, ok := s.GetTorrent(infoHash)
|
||||
if !ok {
|
||||
return fmt.Errorf("torrent not found: %s", infoHash)
|
||||
}
|
||||
|
||||
// Find the largest video file
|
||||
var bestFile *torrent.File
|
||||
var bestIdx int
|
||||
for i, f := range t.Files() {
|
||||
if isVideoFile(f.Path()) {
|
||||
if bestFile == nil || f.Length() > bestFile.Length() {
|
||||
bestFile = f
|
||||
bestIdx = i
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if bestFile == nil {
|
||||
return fmt.Errorf("no video file found in torrent")
|
||||
}
|
||||
|
||||
return s.StreamFile(w, r, infoHash, bestIdx)
|
||||
}
|
||||
|
||||
// GetStreamInfo returns info about an active torrent
|
||||
func (s *Service) GetStreamInfo(infoHash string) (*StreamInfo, error) {
|
||||
t, ok := s.GetTorrent(infoHash)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("torrent not found: %s", infoHash)
|
||||
}
|
||||
return s.getStreamInfo(t), nil
|
||||
}
|
||||
|
||||
func (s *Service) getStreamInfo(t *torrent.Torrent) *StreamInfo {
|
||||
info := &StreamInfo{
|
||||
InfoHash: t.InfoHash().HexString(),
|
||||
Name: t.Name(),
|
||||
Peers: t.Stats().ActivePeers,
|
||||
}
|
||||
|
||||
var totalLength, completed int64
|
||||
for i, f := range t.Files() {
|
||||
info.Files = append(info.Files, FileInfo{
|
||||
Index: i,
|
||||
Path: f.Path(),
|
||||
Size: f.Length(),
|
||||
})
|
||||
totalLength += f.Length()
|
||||
completed += f.BytesCompleted()
|
||||
}
|
||||
|
||||
info.Size = totalLength
|
||||
if totalLength > 0 {
|
||||
info.Progress = float64(completed) / float64(totalLength) * 100
|
||||
}
|
||||
|
||||
stats := t.Stats()
|
||||
info.DownloadRate = stats.ConnStats.BytesReadData.Int64()
|
||||
|
||||
return info
|
||||
}
|
||||
|
||||
// DropTorrent removes a torrent from the client
|
||||
func (s *Service) DropTorrent(infoHash string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if t, ok := s.activeTorrents[infoHash]; ok {
|
||||
t.Drop()
|
||||
delete(s.activeTorrents, infoHash)
|
||||
}
|
||||
}
|
||||
|
||||
// Close shuts down the torrent client
|
||||
func (s *Service) Close() {
|
||||
s.mu.Lock()
|
||||
for _, t := range s.activeTorrents {
|
||||
t.Drop()
|
||||
}
|
||||
s.activeTorrents = nil
|
||||
s.mu.Unlock()
|
||||
|
||||
if s.hls != nil {
|
||||
s.hls.Shutdown()
|
||||
}
|
||||
|
||||
s.client.Close()
|
||||
}
|
||||
|
||||
func isVideoFile(path string) bool {
|
||||
videoExts := []string{".mp4", ".mkv", ".avi", ".mov", ".webm", ".wmv", ".flv"}
|
||||
lower := strings.ToLower(path)
|
||||
for _, ext := range videoExts {
|
||||
if strings.HasSuffix(lower, ext) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// ParseMagnetHash extracts info hash from a magnet URI
|
||||
func ParseMagnetHash(magnetURI string) (string, error) {
|
||||
spec, err := torrent.TorrentSpecFromMagnetUri(magnetURI)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return spec.InfoHash.HexString(), nil
|
||||
}
|
||||
|
||||
// MagnetFromHash creates a minimal magnet URI from an info hash
|
||||
func MagnetFromHash(infoHash string) (string, error) {
|
||||
var ih metainfo.Hash
|
||||
if err := ih.FromHexString(infoHash); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return fmt.Sprintf("magnet:?xt=urn:btih:%s", ih.HexString()), nil
|
||||
}
|
||||
|
||||
// GetVideoFilePath returns the filesystem path to the main video file
|
||||
func (s *Service) GetVideoFilePath(infoHash string) (string, error) {
|
||||
t, ok := s.GetTorrent(infoHash)
|
||||
if !ok {
|
||||
return "", fmt.Errorf("torrent not found: %s", infoHash)
|
||||
}
|
||||
|
||||
// Find the largest video file
|
||||
var bestFile *torrent.File
|
||||
for _, f := range t.Files() {
|
||||
if isVideoFile(f.Path()) {
|
||||
if bestFile == nil || f.Length() > bestFile.Length() {
|
||||
bestFile = f
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if bestFile == nil {
|
||||
return "", fmt.Errorf("no video file found in torrent")
|
||||
}
|
||||
|
||||
// Return path relative to data dir
|
||||
return filepath.Join("/tmp", "mal-streams", bestFile.Path()), nil
|
||||
}
|
||||
|
||||
// StartHLS starts HLS transcoding for a torrent
|
||||
func (s *Service) StartHLS(ctx context.Context, infoHash string) (*HLSSession, error) {
|
||||
if s.hls == nil {
|
||||
return nil, fmt.Errorf("HLS transcoding not available (ffmpeg not found)")
|
||||
}
|
||||
|
||||
// Check if session already exists
|
||||
if session, ok := s.hls.GetSession(infoHash); ok {
|
||||
return session, nil
|
||||
}
|
||||
|
||||
// Get torrent and video file
|
||||
t, ok := s.GetTorrent(infoHash)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("torrent not found: %s", infoHash)
|
||||
}
|
||||
|
||||
// Find the largest video file
|
||||
var videoFile *torrent.File
|
||||
for _, f := range t.Files() {
|
||||
if isVideoFile(f.Path()) {
|
||||
if videoFile == nil || f.Length() > videoFile.Length() {
|
||||
videoFile = f
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if videoFile == nil {
|
||||
return nil, fmt.Errorf("no video file found in torrent")
|
||||
}
|
||||
|
||||
// Prioritize downloading the beginning of the file for ffmpeg
|
||||
videoFile.Download()
|
||||
reader := videoFile.NewReader()
|
||||
reader.SetReadahead(10 * 1024 * 1024) // 10MB readahead
|
||||
reader.SetResponsive()
|
||||
|
||||
// Wait for at least 2MB to be available before starting ffmpeg
|
||||
minBytes := int64(2 * 1024 * 1024)
|
||||
s.logger.Info("waiting for initial data", "hash", infoHash, "need", minBytes)
|
||||
|
||||
ticker := time.NewTicker(500 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
|
||||
timeout := time.After(60 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-timeout:
|
||||
return nil, fmt.Errorf("timeout waiting for video data")
|
||||
case <-ticker.C:
|
||||
completed := videoFile.BytesCompleted()
|
||||
if completed >= minBytes {
|
||||
s.logger.Info("got enough data, starting transcoding", "hash", infoHash, "bytes", completed)
|
||||
goto ready
|
||||
}
|
||||
s.logger.Debug("waiting for data", "hash", infoHash, "completed", completed, "need", minBytes)
|
||||
}
|
||||
}
|
||||
|
||||
ready:
|
||||
// Start transcoding with the reader piped to ffmpeg
|
||||
session, err := s.hls.StartSessionWithReader(infoHash, reader)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Wait for first segments to be ready
|
||||
waitCtx, cancel := context.WithTimeout(ctx, 60*time.Second)
|
||||
defer cancel()
|
||||
|
||||
if err := session.WaitReady(waitCtx); err != nil {
|
||||
s.hls.StopSession(infoHash)
|
||||
return nil, fmt.Errorf("HLS not ready: %w", err)
|
||||
}
|
||||
|
||||
return session, nil
|
||||
}
|
||||
|
||||
// GetHLSSession returns an existing HLS session
|
||||
func (s *Service) GetHLSSession(infoHash string) (*HLSSession, bool) {
|
||||
if s.hls == nil {
|
||||
return nil, false
|
||||
}
|
||||
return s.hls.GetSession(infoHash)
|
||||
}
|
||||
|
||||
// HasHLS returns whether HLS transcoding is available
|
||||
func (s *Service) HasHLS() bool {
|
||||
return s.hls != nil
|
||||
}
|
||||
Reference in New Issue
Block a user