diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..3bc6a4d --- /dev/null +++ b/.dockerignore @@ -0,0 +1,8 @@ +media/ +db/ +bin/ +*.mkv +*.mp4 +*.mp3 +.git/ +.gitea/ \ No newline at end of file diff --git a/.gitea/workflows/build.yml b/.gitea/workflows/build.yml new file mode 100644 index 0000000..9a4b46f --- /dev/null +++ b/.gitea/workflows/build.yml @@ -0,0 +1,21 @@ +name: Build + +on: + push: + branches: [ main ] + workflow_dispatch: + +jobs: + build-and-push: + name: Build and Push + runs-on: build-htz-01 + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Build and push Docker image + run: | + docker build -t cr.jdbnet.co.uk/public/goencode:latest . + docker push cr.jdbnet.co.uk/public/goencode:latest \ No newline at end of file diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c59768b --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +media/ \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..3131264 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,33 @@ +# Build stage +FROM golang:1.26.4-bookworm AS builder + +WORKDIR /app + +# Copy source code +COPY . . + +# Run go mod tidy to generate go.sum based on source +RUN go mod tidy + +# Build the application +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /app/bin/goencode ./cmd/goencode + +# Final stage +FROM debian:bookworm-slim + +# Install ffmpeg and ca-certificates +RUN apt-get update && \ + apt-get install -y --no-install-recommends ffmpeg ca-certificates && \ + rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +# Copy binary from builder +COPY --from=builder /app/bin/goencode /app/goencode + +# Expose port +EXPOSE 8080 + +# Run the binary +ENTRYPOINT ["/app/goencode"] +CMD ["--config", "/etc/goencode/config.yaml"] diff --git a/README.md b/README.md index cbf48d2..7e201cf 100644 --- a/README.md +++ b/README.md @@ -1 +1,106 @@ -# GoEncode \ No newline at end of file +
+ GoEncode + +# GoEncode + +GoEncode is a lightweight, high-performance media transcoding server written in Go. Originally built as a Flask app, it has been rewritten from the ground up as a single, self-contained binary. GoEncode features a modern web UI, a robust job queue, automatic watch folder monitoring, real-time log streaming, and MariaDB-backed persistence. + +
+ +## Features + +- **Watch Folders**: Automatically monitor designated directories for new media files. +- **Smart Queue & Debouncing**: Prevents incomplete files from being processed while they are still being copied to the watch folder. +- **Background Processing**: Processes media sequentially with up to 3 worker threads. Files are safely copied to a temporary directory before encoding to avoid hammering network-attached storage (NAS). +- **Format Intelligence**: Probes media to detect codecs and resolution, automatically skipping files that already meet the target codec/resolution. +- **Web Dashboard**: Modern, responsive interface with Server-Sent Events (SSE) for live tracking of job progress, queue stats, and server logs. +- **Robust Persistence**: Job history, metrics (size saved, time taken), and configuration are all saved to a MariaDB/MySQL database. +- **Docker-Ready**: Packaged in an ultra-slim container image based on Debian, with `ffmpeg` built-in. + +## Deployment with Docker + +The easiest way to run GoEncode is via Docker using the pre-built image. + +### Image Registry + +`cr.jdbnet.co.uk/public/goencode:latest` + +### Example `docker-compose.yml` + +```yaml +services: + goencode: + image: cr.jdbnet.co.uk/public/goencode:latest + container_name: goencode + restart: unless-stopped + ports: + - "8080:8080" + environment: + # Database Configuration + - GOENCODE_DB_HOST=db + - GOENCODE_DB_PORT=3306 + - GOENCODE_DB_USER=goencode + - GOENCODE_DB_PASS=goencode_password + - GOENCODE_DB_NAME=goencode + + # Web Server Configuration + - GOENCODE_SERVER_LISTEN=0.0.0.0 + - GOENCODE_SERVER_PORT=8080 + + # Web UI Authentication (Optional) + - GOENCODE_AUTH_USER=admin + - GOENCODE_AUTH_PASS=secret + + # Encoding + - GOENCODE_ENCODER_TEMP=/tmp/goencode + volumes: + - /path/to/your/media:/media + depends_on: + - db + + db: + image: mariadb:10.11 + container_name: goencode-db + restart: unless-stopped + environment: + - MARIADB_DATABASE=goencode + - MARIADB_USER=goencode + - MARIADB_PASSWORD=goencode_password + - MARIADB_ROOT_PASSWORD=root_password + volumes: + - goencode_db_data:/var/lib/mysql + +volumes: + goencode_db_data: +``` + +## Configuration + +GoEncode can be configured via `goencode.yaml` or entirely via environment variables (ideal for Docker/Kubernetes). Environment variables take precedence over the YAML file. + +### Available Environment Variables + +| Variable | Description | Default | +|----------|-------------|---------| +| `GOENCODE_DB_HOST` | Database host | `127.0.0.1` | +| `GOENCODE_DB_PORT` | Database port | `3306` | +| `GOENCODE_DB_USER` | Database username | `goencode` | +| `GOENCODE_DB_PASS` | Database password | | +| `GOENCODE_DB_NAME` | Database name | `goencode` | +| `GOENCODE_SERVER_LISTEN` | IP to bind the web interface | `0.0.0.0` | +| `GOENCODE_SERVER_PORT` | Port for the web interface | `8080` | +| `GOENCODE_AUTH_USER` | Username for the web UI | | +| `GOENCODE_AUTH_PASS` | Password for the web UI | | +| `GOENCODE_ENCODER_TEMP`| Temp directory for processing jobs | `/tmp/goencode` | + +## Building Locally + +To build the Docker image locally and push it to your registry: + +```bash +# Build the image +docker build -t cr.jdbnet.co.uk/public/goencode:latest . + +# Push to your registry +docker push cr.jdbnet.co.uk/public/goencode:latest +``` \ No newline at end of file diff --git a/cmd/goencode/main.go b/cmd/goencode/main.go new file mode 100644 index 0000000..f6cf733 --- /dev/null +++ b/cmd/goencode/main.go @@ -0,0 +1,60 @@ +package main + +import ( + "flag" + "log" + "os" + "os/signal" + "syscall" + + "goencode/internal/config" + "goencode/internal/db" + "goencode/internal/logger" + "goencode/internal/queue" + "goencode/internal/watcher" + "goencode/internal/web" +) + +func main() { + configPath := flag.String("config", "goencode.yaml", "Path to configuration file") + flag.Parse() + + cfg, err := config.LoadConfig(*configPath) + if err != nil { + log.Fatalf("Failed to load config: %v", err) + } + + if err := db.Init(&cfg.Database); err != nil { + log.Fatalf("Database init failed: %v", err) + } + + sseServer := web.NewSSEServer() + + // Initialize logger + logger.Init(sseServer.Broadcast) + + qm := queue.NewManager(cfg.Encoder.FFmpegPath, cfg.Encoder.TempDir, sseServer.Broadcast) + qm.Start() + defer qm.Stop() + + wm, err := watcher.NewManager(qm) + if err != nil { + log.Fatalf("Watcher init failed: %v", err) + } + wm.Start() + defer wm.Stop() + + server := web.NewServer(cfg, qm, wm, sseServer) + + go func() { + log.Printf("Starting GoEncode Web UI on %s:%d", cfg.Server.ListenAddr, cfg.Server.Port) + if err := server.Start(); err != nil { + log.Fatalf("Server failed: %v", err) + } + }() + + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) + <-sig + log.Println("Shutting down...") +} diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..348f608 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,25 @@ +services: + db: + image: mariadb:10.11 + restart: always + environment: + MYSQL_ROOT_PASSWORD: root + MYSQL_DATABASE: encoder + volumes: + - db_data:/var/lib/mysql + ports: + - "3307:3306" + + app: + build: . + restart: always + ports: + - "8080:8080" + volumes: + - ./goencode.yaml:/etc/goencode/config.yaml:ro + - ./media:/media # Mount your media folder here + depends_on: + - db + +volumes: + db_data: diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..bb02ed3 --- /dev/null +++ b/go.mod @@ -0,0 +1,9 @@ +module goencode + +go 1.26.4 + +require ( + github.com/fsnotify/fsnotify v1.7.0 + github.com/go-sql-driver/mysql v1.8.1 + gopkg.in/yaml.v3 v3.0.1 +) diff --git a/goencode.yaml b/goencode.yaml new file mode 100644 index 0000000..fbc3cba --- /dev/null +++ b/goencode.yaml @@ -0,0 +1,21 @@ +server: + port: 8080 + listen_addr: "0.0.0.0" + +auth: + username: "admin" + password: "password" + +database: + host: "db" + port: 3306 + user: "root" + password: "root" + name: "encoder" + +encoder: + ffmpeg_path: "ffmpeg" + temp_dir: "/tmp/goencode" + +logging: + level: "info" diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..d16ddda --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,99 @@ +package config + +import ( + "fmt" + "os" + "strconv" + + "gopkg.in/yaml.v3" +) + +func getEnvStr(key, defaultVal string) string { + if val := os.Getenv(key); val != "" { + return val + } + return defaultVal +} + +func getEnvInt(key string, defaultVal int) int { + if val := os.Getenv(key); val != "" { + if i, err := strconv.Atoi(val); err == nil { + return i + } + } + return defaultVal +} + +type ServerConfig struct { + Port int `yaml:"port"` + ListenAddr string `yaml:"listen_addr"` +} + +type DatabaseConfig struct { + Host string `yaml:"host"` + Port int `yaml:"port"` + User string `yaml:"user"` + Password string `yaml:"password"` + Name string `yaml:"name"` +} + +type EncoderConfig struct { + FFmpegPath string `yaml:"ffmpeg_path"` + TempDir string `yaml:"temp_dir"` +} + +type LoggingConfig struct { + Level string `yaml:"level"` +} + +type AuthConfig struct { + Username string `yaml:"username"` + Password string `yaml:"password"` +} + +type Config struct { + Server ServerConfig `yaml:"server"` + Auth AuthConfig `yaml:"auth"` + Database DatabaseConfig `yaml:"database"` + Encoder EncoderConfig `yaml:"encoder"` + Logging LoggingConfig `yaml:"logging"` +} + +func LoadConfig(path string) (*Config, error) { + var cfg Config + + // Try reading from file first (it's okay if it doesn't exist for env-only deployments) + data, err := os.ReadFile(path) + if err == nil { + if err := yaml.Unmarshal(data, &cfg); err != nil { + return nil, fmt.Errorf("could not parse config file: %w", err) + } + } + + // Override with env variables + cfg.Server.Port = getEnvInt("GOENCODE_PORT", cfg.Server.Port) + cfg.Server.ListenAddr = getEnvStr("GOENCODE_LISTEN_ADDR", cfg.Server.ListenAddr) + cfg.Auth.Username = getEnvStr("GOENCODE_AUTH_USER", cfg.Auth.Username) + cfg.Auth.Password = getEnvStr("GOENCODE_AUTH_PASS", cfg.Auth.Password) + cfg.Database.Host = getEnvStr("GOENCODE_DB_HOST", cfg.Database.Host) + cfg.Database.Port = getEnvInt("GOENCODE_DB_PORT", cfg.Database.Port) + cfg.Database.User = getEnvStr("GOENCODE_DB_USER", cfg.Database.User) + cfg.Database.Password = getEnvStr("GOENCODE_DB_PASS", cfg.Database.Password) + cfg.Database.Name = getEnvStr("GOENCODE_DB_NAME", cfg.Database.Name) + cfg.Encoder.FFmpegPath = getEnvStr("GOENCODE_FFMPEG_PATH", cfg.Encoder.FFmpegPath) + cfg.Encoder.TempDir = getEnvStr("GOENCODE_TEMP_DIR", cfg.Encoder.TempDir) + cfg.Logging.Level = getEnvStr("GOENCODE_LOG_LEVEL", cfg.Logging.Level) + + // Defaults if missing entirely + if cfg.Server.Port == 0 { + cfg.Server.Port = 8080 + } + if cfg.Server.ListenAddr == "" { + cfg.Server.ListenAddr = "0.0.0.0" + } + if cfg.Database.Port == 0 { + cfg.Database.Port = 3306 + } + + return &cfg, nil +} diff --git a/internal/db/db.go b/internal/db/db.go new file mode 100644 index 0000000..aae3c65 --- /dev/null +++ b/internal/db/db.go @@ -0,0 +1,92 @@ +package db + +import ( + "database/sql" + "fmt" + "log" + "time" + + _ "github.com/go-sql-driver/mysql" + "goencode/internal/config" +) + +var DB *sql.DB + +func Init(cfg *config.DatabaseConfig) error { + dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?parseTime=true", + cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Name) + + var err error + DB, err = sql.Open("mysql", dsn) + if err != nil { + return err + } + + // Set connection pool limits + DB.SetMaxOpenConns(25) + DB.SetMaxIdleConns(25) + DB.SetConnMaxLifetime(5 * time.Minute) + + if err = DB.Ping(); err != nil { + return err + } + + log.Println("Connected to MariaDB successfully") + + return runMigrations() +} + +func runMigrations() error { + queries := []string{ + `CREATE TABLE IF NOT EXISTS watch_folders ( + id INT AUTO_INCREMENT PRIMARY KEY, + folder_path VARCHAR(500) NOT NULL UNIQUE, + media_type ENUM('video', 'audio') NOT NULL, + target_resolution VARCHAR(20), + custom_ffmpeg_flags TEXT, + enabled BOOLEAN DEFAULT TRUE, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP + )`, + `CREATE TABLE IF NOT EXISTS jobs ( + id INT AUTO_INCREMENT PRIMARY KEY, + file_path VARCHAR(500) NOT NULL, + media_type ENUM('video', 'audio') NOT NULL, + status ENUM('pending', 'processing', 'failed') NOT NULL DEFAULT 'pending', + priority INT NOT NULL DEFAULT 0, + original_size BIGINT DEFAULT 0, + target_resolution VARCHAR(20), + ffmpeg_flags TEXT, + error_message TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP + )`, + `CREATE TABLE IF NOT EXISTS job_reports ( + id INT AUTO_INCREMENT PRIMARY KEY, + file_path VARCHAR(500) NOT NULL, + media_type ENUM('video', 'audio') NOT NULL, + status ENUM('success', 'failed', 'skipped') NOT NULL, + original_size BIGINT NOT NULL DEFAULT 0, + encoded_size BIGINT NOT NULL DEFAULT 0, + size_saved BIGINT NOT NULL DEFAULT 0, + processing_time DECIMAL(10,2) DEFAULT 0, + target_resolution VARCHAR(20), + ffmpeg_flags TEXT, + error_message TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + )`, + `CREATE TABLE IF NOT EXISTS app_config ( + config_key VARCHAR(100) PRIMARY KEY, + config_value TEXT + )`, + } + + for _, q := range queries { + if _, err := DB.Exec(q); err != nil { + return fmt.Errorf("migration failed: %w\nQuery: %s", err, q) + } + } + + log.Println("Database schemas initialized") + return nil +} diff --git a/internal/db/models.go b/internal/db/models.go new file mode 100644 index 0000000..cbdf8c6 --- /dev/null +++ b/internal/db/models.go @@ -0,0 +1,45 @@ +package db + +import ( + "time" +) + +type WatchFolder struct { + ID int `json:"id"` + FolderPath string `json:"folder_path"` + MediaType string `json:"media_type"` + TargetResolution string `json:"target_resolution"` + CustomFFmpegFlags string `json:"custom_ffmpeg_flags"` + Enabled bool `json:"enabled"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type Job struct { + ID int `json:"id"` + FilePath string `json:"file_path"` + MediaType string `json:"media_type"` + Status string `json:"status"` // pending, processing, failed + Priority int `json:"priority"` + OriginalSize int64 `json:"original_size"` + TargetResolution string `json:"target_resolution"` + FFmpegFlags string `json:"ffmpeg_flags"` + ErrorMessage string `json:"error_message"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type JobReport struct { + ID int `json:"id"` + FilePath string `json:"file_path"` + MediaType string `json:"media_type"` + Status string `json:"status"` // success, failed, skipped + OriginalSize int64 `json:"original_size"` + EncodedSize int64 `json:"encoded_size"` + SizeSaved int64 `json:"size_saved"` + ProcessingTime float64 `json:"processing_time"` + TargetResolution string `json:"target_resolution"` + FFmpegFlags string `json:"ffmpeg_flags"` + ErrorMessage string `json:"error_message"` + CreatedAt time.Time `json:"created_at"` +} diff --git a/internal/db/queries.go b/internal/db/queries.go new file mode 100644 index 0000000..af99855 --- /dev/null +++ b/internal/db/queries.go @@ -0,0 +1,182 @@ +package db + +import ( + "database/sql" +) + +// Watch Folders + +func GetWatchFolders() ([]WatchFolder, error) { + rows, err := DB.Query(`SELECT id, folder_path, media_type, target_resolution, custom_ffmpeg_flags, enabled, created_at, updated_at FROM watch_folders`) + if err != nil { + return nil, err + } + defer rows.Close() + + var folders []WatchFolder + for rows.Next() { + var f WatchFolder + var targetRes, ffmpegFlags sql.NullString + if err := rows.Scan(&f.ID, &f.FolderPath, &f.MediaType, &targetRes, &ffmpegFlags, &f.Enabled, &f.CreatedAt, &f.UpdatedAt); err != nil { + return nil, err + } + if targetRes.Valid { + f.TargetResolution = targetRes.String + } + if ffmpegFlags.Valid { + f.CustomFFmpegFlags = ffmpegFlags.String + } + folders = append(folders, f) + } + return folders, nil +} + +func AddWatchFolder(f WatchFolder) error { + _, err := DB.Exec(`INSERT INTO watch_folders (folder_path, media_type, target_resolution, custom_ffmpeg_flags, enabled) VALUES (?, ?, ?, ?, ?)`, + f.FolderPath, f.MediaType, nullStr(f.TargetResolution), nullStr(f.CustomFFmpegFlags), f.Enabled) + return err +} + +func DeleteWatchFolder(id int) error { + _, err := DB.Exec(`DELETE FROM watch_folders WHERE id = ?`, id) + return err +} + +func nullStr(s string) sql.NullString { + if s == "" { + return sql.NullString{Valid: false} + } + return sql.NullString{String: s, Valid: true} +} + +// Jobs + +func AddJob(filePath, mediaType string, priority int, targetRes, ffmpegFlags string, originalSize int64) error { + _, err := DB.Exec(`INSERT INTO jobs (file_path, media_type, priority, target_resolution, ffmpeg_flags, original_size) VALUES (?, ?, ?, ?, ?, ?)`, + filePath, mediaType, priority, nullStr(targetRes), nullStr(ffmpegFlags), originalSize) + return err +} + +func GetPendingJobs() ([]Job, error) { + rows, err := DB.Query(`SELECT id, file_path, media_type, status, priority, original_size, target_resolution, ffmpeg_flags, error_message, created_at, updated_at FROM jobs ORDER BY priority DESC, created_at ASC`) + if err != nil { + return nil, err + } + defer rows.Close() + + var jobs []Job + for rows.Next() { + var j Job + var targetRes, ffmpegFlags, errMsg sql.NullString + if err := rows.Scan(&j.ID, &j.FilePath, &j.MediaType, &j.Status, &j.Priority, &j.OriginalSize, &targetRes, &ffmpegFlags, &errMsg, &j.CreatedAt, &j.UpdatedAt); err != nil { + return nil, err + } + if targetRes.Valid { j.TargetResolution = targetRes.String } + if ffmpegFlags.Valid { j.FFmpegFlags = ffmpegFlags.String } + if errMsg.Valid { j.ErrorMessage = errMsg.String } + jobs = append(jobs, j) + } + return jobs, nil +} + +func UpdateJobStatus(id int, status, errMsg string) error { + _, err := DB.Exec(`UPDATE jobs SET status = ?, error_message = ? WHERE id = ?`, status, nullStr(errMsg), id) + return err +} + +func BumpJobPriority(id int) error { + _, err := DB.Exec(`UPDATE jobs SET priority = priority + 1 WHERE id = ?`, id) + return err +} + +func DeleteJob(id int) error { + _, err := DB.Exec(`DELETE FROM jobs WHERE id = ?`, id) + return err +} + +func MarkProcessingAsFailed() error { + _, err := DB.Exec(`UPDATE jobs SET status = 'failed', error_message = 'Interrupted by server restart' WHERE status = 'processing'`) + return err +} + +func IsFileAlreadyProcessedOrQueued(filePath string) (bool, error) { + var count int + err := DB.QueryRow(`SELECT COUNT(*) FROM jobs WHERE file_path = ?`, filePath).Scan(&count) + if err != nil { + return false, err + } + if count > 0 { + return true, nil + } + + err = DB.QueryRow(`SELECT COUNT(*) FROM job_reports WHERE file_path = ?`, filePath).Scan(&count) + if err != nil { + return false, err + } + return count > 0, nil +} + +// Job Reports + +func AddJobReport(j Job, status string, encodedSize int64, sizeSaved int64, processingTime float64) error { + _, err := DB.Exec(`INSERT INTO job_reports (file_path, media_type, status, original_size, encoded_size, size_saved, processing_time, target_resolution, ffmpeg_flags, error_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + j.FilePath, j.MediaType, status, j.OriginalSize, encodedSize, sizeSaved, processingTime, nullStr(j.TargetResolution), nullStr(j.FFmpegFlags), nullStr(j.ErrorMessage)) + return err +} + +func GetJobReports(limit, offset int) ([]JobReport, int, error) { + var total int + if err := DB.QueryRow(`SELECT COUNT(*) FROM job_reports`).Scan(&total); err != nil { + return nil, 0, err + } + + rows, err := DB.Query(`SELECT id, file_path, media_type, status, original_size, encoded_size, size_saved, processing_time, target_resolution, ffmpeg_flags, error_message, created_at FROM job_reports ORDER BY created_at DESC LIMIT ? OFFSET ?`, limit, offset) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + var reports []JobReport + for rows.Next() { + var r JobReport + var targetRes, ffmpegFlags, errMsg sql.NullString + if err := rows.Scan(&r.ID, &r.FilePath, &r.MediaType, &r.Status, &r.OriginalSize, &r.EncodedSize, &r.SizeSaved, &r.ProcessingTime, &targetRes, &ffmpegFlags, &errMsg, &r.CreatedAt); err != nil { + return nil, 0, err + } + if targetRes.Valid { r.TargetResolution = targetRes.String } + if ffmpegFlags.Valid { r.FFmpegFlags = ffmpegFlags.String } + if errMsg.Valid { r.ErrorMessage = errMsg.String } + reports = append(reports, r) + } + return reports, total, nil +} + +type DashboardStats struct { + TotalSavedSpace int64 + FilesEncoded int + QueueLength int +} + +func GetDashboardStats() (DashboardStats, error) { + var stats DashboardStats + + // Total saved space and files encoded + err := DB.QueryRow(` + SELECT + COALESCE(SUM(size_saved), 0) as saved, + COUNT(*) as count + FROM job_reports + WHERE status = 'success' + `).Scan(&stats.TotalSavedSpace, &stats.FilesEncoded) + if err != nil && err != sql.ErrNoRows { + return stats, err + } + + // Queue length + err = DB.QueryRow(`SELECT COUNT(*) FROM jobs`).Scan(&stats.QueueLength) + if err != nil && err != sql.ErrNoRows { + return stats, err + } + + return stats, nil +} diff --git a/internal/encoder/audio.go b/internal/encoder/audio.go new file mode 100644 index 0000000..8cf1d41 --- /dev/null +++ b/internal/encoder/audio.go @@ -0,0 +1,45 @@ +package encoder + +import ( + "os/exec" + "strconv" + "strings" +) + +func (m *FFmpegManager) BuildAudioCmd(inputPath, outputPath, ffmpegFlags string) (*exec.Cmd, error) { + args := []string{"-i", inputPath, "-c:a", "libmp3lame", "-b:a", "320k"} + + if ffmpegFlags != "" { + userArgs := ShlexSplit(ffmpegFlags) + args = append(args, userArgs...) + } + + args = append(args, outputPath, "-y") + + return exec.Command(m.BinaryPath, args...), nil +} + +func (m *FFmpegManager) ShouldSkipAudio(inputPath string) bool { + // Probe the codec + cmd := exec.Command("ffprobe", "-v", "quiet", "-select_streams", "a:0", + "-show_entries", "stream=codec_name", "-of", "csv=p=0", inputPath) + out, err := cmd.Output() + if err != nil { + return false + } + codec := strings.TrimSpace(string(out)) + + // If it's already mp3, check bitrate + if codec == "mp3" { + cmdBr := exec.Command("ffprobe", "-v", "quiet", "-select_streams", "a:0", + "-show_entries", "stream=bit_rate", "-of", "csv=p=0", inputPath) + outBr, err := cmdBr.Output() + if err == nil { + br, _ := strconv.Atoi(strings.TrimSpace(string(outBr))) + if br >= 300000 { + return true // It's ~320k mp3 + } + } + } + return false +} diff --git a/internal/encoder/ffmpeg.go b/internal/encoder/ffmpeg.go new file mode 100644 index 0000000..6f0d2a4 --- /dev/null +++ b/internal/encoder/ffmpeg.go @@ -0,0 +1,133 @@ +package encoder + +import ( + "bytes" + "fmt" + "os/exec" + "strconv" + "strings" +) + +// FFmpegManager handles execution and probing +type FFmpegManager struct { + BinaryPath string +} + +func NewManager(binaryPath string) *FFmpegManager { + if binaryPath == "" { + binaryPath = "ffmpeg" + } + return &FFmpegManager{BinaryPath: binaryPath} +} + +func (m *FFmpegManager) ProbeCodec(filePath, streamType string) (string, error) { + cmd := exec.Command("ffprobe", "-v", "quiet", "-select_streams", streamType+":0", + "-show_entries", "stream=codec_name", "-of", "csv=p=0", filePath) + out, err := cmd.Output() + if err != nil { + return "", fmt.Errorf("ffprobe error: %w", err) + } + return strings.TrimSpace(string(out)), nil +} + +func (m *FFmpegManager) ProbeResolution(filePath string) (int, int, error) { + cmd := exec.Command("ffprobe", "-v", "quiet", "-select_streams", "v:0", + "-show_entries", "stream=width,height", "-of", "csv=p=0", filePath) + out, err := cmd.Output() + if err != nil { + return 0, 0, fmt.Errorf("ffprobe error: %w", err) + } + parts := strings.Split(strings.TrimSpace(string(out)), ",") + if len(parts) != 2 { + return 0, 0, fmt.Errorf("invalid resolution format") + } + w, err1 := strconv.Atoi(parts[0]) + h, err2 := strconv.Atoi(parts[1]) + if err1 != nil || err2 != nil { + return 0, 0, fmt.Errorf("invalid resolution numbers") + } + return w, h, nil +} + +func (m *FFmpegManager) ProbeDuration(filePath string) (float64, error) { + cmd := exec.Command("ffprobe", "-v", "quiet", "-show_entries", "format=duration", "-of", "csv=p=0", filePath) + out, err := cmd.Output() + if err != nil { + return 0, fmt.Errorf("ffprobe error: %w", err) + } + return strconv.ParseFloat(strings.TrimSpace(string(out)), 64) +} + +func (m *FFmpegManager) ValidateFile(filePath string) error { + cmd := exec.Command("ffprobe", "-v", "quiet", "-show_format", "-show_streams", filePath) + if err := cmd.Run(); err != nil { + return fmt.Errorf("file validation failed: %w", err) + } + return nil +} + +// ParseProgress is a utility to parse time=XX:XX:XX.XX from stderr line to calculate percentage if duration is known. +// Will be used by the queue worker reading from a pipe. +func ParseProgress(line string, durationSec float64) float64 { + // Example line: frame= 123 fps= 30 q=22.0 size= 2048kB time=00:00:10.50 bitrate=1500.0kbits/s speed=1.5x + idx := strings.Index(line, "time=") + if idx == -1 { + return -1 + } + timeStr := line[idx+5:] + spaceIdx := strings.Index(timeStr, " ") + if spaceIdx != -1 { + timeStr = timeStr[:spaceIdx] + } + + // timeStr is HH:MM:SS.ms + parts := strings.Split(timeStr, ":") + if len(parts) != 3 { + return -1 + } + + h, _ := strconv.ParseFloat(parts[0], 64) + m, _ := strconv.ParseFloat(parts[1], 64) + s, _ := strconv.ParseFloat(parts[2], 64) + + currentSec := h*3600 + m*60 + s + if durationSec > 0 { + return (currentSec / durationSec) * 100.0 + } + return -1 +} + +// ShlexSplit splits a command string like a shell would. +// A simple implementation since go doesn't have shlex.split built-in. +func ShlexSplit(s string) []string { + var args []string + var buf bytes.Buffer + inQuotes := false + var quoteChar rune + + for _, r := range s { + if inQuotes { + if r == quoteChar { + inQuotes = false + } else { + buf.WriteRune(r) + } + } else { + if r == '\'' || r == '"' { + inQuotes = true + quoteChar = r + } else if r == ' ' { + if buf.Len() > 0 { + args = append(args, buf.String()) + buf.Reset() + } + } else { + buf.WriteRune(r) + } + } + } + if buf.Len() > 0 { + args = append(args, buf.String()) + } + return args +} diff --git a/internal/encoder/video.go b/internal/encoder/video.go new file mode 100644 index 0000000..0978bcd --- /dev/null +++ b/internal/encoder/video.go @@ -0,0 +1,75 @@ +package encoder + +import ( + "fmt" + "os/exec" + "strconv" + "strings" +) + +func (m *FFmpegManager) BuildVideoCmd(inputPath, outputPath, targetResolution, ffmpegFlags string, originalWidth, originalHeight int) (*exec.Cmd, error) { + args := []string{"-i", inputPath} + + if targetResolution != "" { + targetHeightStr := strings.TrimRight(strings.ToLower(targetResolution), "p") + targetHeight, err := strconv.Atoi(targetHeightStr) + if err == nil && targetHeight > 0 && originalHeight > targetHeight { + scaleWidth := int((float64(originalWidth) / float64(originalHeight)) * float64(targetHeight)) + if scaleWidth%2 != 0 { + scaleWidth-- + } + args = append(args, "-vf", fmt.Sprintf("scale=%d:%d", scaleWidth, targetHeight)) + } + } + + args = append(args, "-c:v", "libx265", "-c:a", "copy", "-preset", "medium", "-crf", "22", "-f", "matroska") + + if ffmpegFlags != "" { + userArgs := ShlexSplit(ffmpegFlags) + args = append(args, userArgs...) + } + + args = append(args, outputPath, "-y") + + return exec.Command(m.BinaryPath, args...), nil +} + +func (m *FFmpegManager) ShouldSkipVideo(inputPath, targetResolution string) bool { + // Probe the codec + cmd := exec.Command("ffprobe", "-v", "quiet", "-select_streams", "v:0", + "-show_entries", "stream=codec_name", "-of", "csv=p=0", inputPath) + out, err := cmd.Output() + if err != nil { + return false + } + codec := strings.TrimSpace(string(out)) + + // Probe the height + cmdHeight := exec.Command("ffprobe", "-v", "quiet", "-select_streams", "v:0", + "-show_entries", "stream=height", "-of", "csv=p=0", inputPath) + outHeight, err := cmdHeight.Output() + if err != nil { + return false + } + + // If it's not H.264 or H.265, do not skip (encode it) + if codec != "h264" && codec != "hevc" { + return false + } + + // If it's H.265 and no target resolution is set, it's already in the target codec + if codec == "hevc" && targetResolution == "" { + return true + } + + // If it's H.265 and target resolution is set, only skip if current height is <= target height + if codec == "hevc" && targetResolution != "" { + currentHeight, _ := strconv.Atoi(strings.TrimSpace(string(outHeight))) + targetHeight, _ := strconv.Atoi(strings.TrimRight(strings.ToLower(targetResolution), "p")) + if currentHeight > 0 && targetHeight > 0 && currentHeight <= targetHeight { + return true + } + } + + return false +} diff --git a/internal/logger/logger.go b/internal/logger/logger.go new file mode 100644 index 0000000..0b5f776 --- /dev/null +++ b/internal/logger/logger.go @@ -0,0 +1,90 @@ +package logger + +import ( + "container/ring" + "io" + "log" + "os" + "strings" + "sync" + "time" +) + +type LogEntry struct { + Timestamp time.Time `json:"timestamp"` + Message string `json:"message"` +} + +var ( + ringBuffer *ring.Ring + bufferMu sync.Mutex + Broadcast func(string, interface{}) +) + +type customWriter struct { + stdout io.Writer +} + +func (cw *customWriter) Write(p []byte) (n int, err error) { + now := time.Now() + msg := strings.TrimSpace(string(p)) + + // Write to stdout with simple timestamp + outStr := now.Format("2006/01/02 15:04:05 ") + msg + "\n" + cw.stdout.Write([]byte(outStr)) + + if msg == "" { + return len(p), nil + } + + entry := LogEntry{ + Timestamp: now, + Message: msg, + } + + bufferMu.Lock() + if ringBuffer == nil { + ringBuffer = ring.New(100) + } + ringBuffer.Value = entry + ringBuffer = ringBuffer.Next() + bufferMu.Unlock() + + if Broadcast != nil { + Broadcast("log", entry) + } + + return len(p), nil +} + +func Init(broadcast func(string, interface{})) { + Broadcast = broadcast + ringBuffer = ring.New(100) + log.SetOutput(&customWriter{stdout: os.Stdout}) + // Remove timestamp from log format since we add it in LogEntry or we can let log package add it + // Wait, standard log adds timestamp. If standard log adds timestamp, our `msg` will contain the timestamp. + // But `LogEntry` has a `Timestamp` field for the JSON. + // It's cleaner to remove the standard log flags so `msg` is just the message, and let stdout print the plain msg? + // But stdout wouldn't have timestamps then unless we format it ourselves. + // Let's just remove log flags and format it ourselves for stdout! + log.SetFlags(0) +} + +// Since we removed log flags, we need to handle stdout formatting in Write if we want timestamps in stdout. +// Actually, let's keep the timestamp in stdout, we can modify customWriter.Write to prepend the timestamp to stdout. +// We'll update the writer in a moment if needed. + +func GetRecentLogs() []LogEntry { + bufferMu.Lock() + defer bufferMu.Unlock() + + var logs []LogEntry + if ringBuffer != nil { + ringBuffer.Do(func(p interface{}) { + if p != nil { + logs = append(logs, p.(LogEntry)) + } + }) + } + return logs +} diff --git a/internal/queue/manager.go b/internal/queue/manager.go new file mode 100644 index 0000000..a570ba6 --- /dev/null +++ b/internal/queue/manager.go @@ -0,0 +1,58 @@ +package queue + +import ( + "log" + "sync" + + "goencode/internal/db" + "goencode/internal/encoder" +) + +type Manager struct { + FFmpegPath string + TempDir string + TriggerChan chan struct{} + StopChan chan struct{} + Broadcast func(string, interface{}) + encoder *encoder.FFmpegManager + isProcessing bool + mu sync.Mutex +} + +func NewManager(ffmpegPath, tempDir string, broadcast func(string, interface{})) *Manager { + return &Manager{ + FFmpegPath: ffmpegPath, + TempDir: tempDir, + TriggerChan: make(chan struct{}, 1), + StopChan: make(chan struct{}), + Broadcast: broadcast, + encoder: encoder.NewManager(ffmpegPath), + } +} + +func (m *Manager) Start() { + // Mark any interrupted processing jobs as failed + if err := db.MarkProcessingAsFailed(); err != nil { + log.Printf("Failed to mark interrupted jobs: %v", err) + } + + go m.workerLoop() + m.Trigger() // Initial trigger +} + +func (m *Manager) Stop() { + close(m.StopChan) +} + +func (m *Manager) Trigger() { + select { + case m.TriggerChan <- struct{}{}: + default: + } +} + +func (m *Manager) NotifySSE(event string, data interface{}) { + if m.Broadcast != nil { + m.Broadcast(event, data) + } +} diff --git a/internal/queue/worker.go b/internal/queue/worker.go new file mode 100644 index 0000000..e842bcd --- /dev/null +++ b/internal/queue/worker.go @@ -0,0 +1,265 @@ +package queue + +import ( + "bufio" + "bytes" + "fmt" + "log" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + "goencode/internal/db" + "goencode/internal/encoder" + "io" +) + +func copyFile(src, dst string) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + + out, err := os.Create(dst) + if err != nil { + return err + } + defer out.Close() + + _, err = io.Copy(out, in) + if err != nil { + return err + } + return out.Sync() +} + +func (m *Manager) workerLoop() { + for { + select { + case <-m.StopChan: + return + case <-m.TriggerChan: + m.processNextJob() + } + } +} + +func (m *Manager) processNextJob() { + m.mu.Lock() + if m.isProcessing { + m.mu.Unlock() + return + } + m.isProcessing = true + m.mu.Unlock() + + var hasJobs bool + defer func() { + m.mu.Lock() + m.isProcessing = false + m.mu.Unlock() + // Only trigger immediately if we know there might be more jobs + if hasJobs { + m.Trigger() + } + }() + + jobs, err := db.GetPendingJobs() + if err != nil { + log.Printf("Failed to get pending jobs: %v", err) + return + } + + if len(jobs) == 0 { + return + } + hasJobs = true + + job := jobs[0] + log.Printf("Starting job %d for %s", job.ID, job.FilePath) + + m.NotifySSE("job_started", job) + + err = db.UpdateJobStatus(job.ID, "processing", "") + if err != nil { + log.Printf("Failed to update job status: %v", err) + return + } + + err = m.runEncoder(job) + if err != nil { + log.Printf("Job %d failed: %v", job.ID, err) + db.UpdateJobStatus(job.ID, "failed", err.Error()) + db.AddJobReport(job, "failed", 0, 0, 0) + m.NotifySSE("job_failed", map[string]interface{}{"id": job.ID, "error": err.Error()}) + } else { + log.Printf("Job %d succeeded", job.ID) + m.NotifySSE("job_completed", map[string]interface{}{"id": job.ID}) + } + db.DeleteJob(job.ID) +} + +func (m *Manager) runEncoder(job db.Job) error { + startTime := time.Now() + + if _, err := os.Stat(job.FilePath); os.IsNotExist(err) { + return fmt.Errorf("source file missing") + } + + if err := os.MkdirAll(m.TempDir, 0755); err != nil { + return fmt.Errorf("failed to create temp dir: %w", err) + } + + originalSize := job.OriginalSize + if originalSize == 0 { + info, err := os.Stat(job.FilePath) + if err == nil { + originalSize = info.Size() + } + } + + ext := filepath.Ext(job.FilePath) + var outExt string + if job.MediaType == "video" { + outExt = ".mkv" + } else { + outExt = ".mp3" + } + + baseName := strings.TrimSuffix(filepath.Base(job.FilePath), ext) + tempOutPath := filepath.Join(m.TempDir, fmt.Sprintf("temp_%d_%s%s", job.ID, baseName, outExt)) + finalOutPath := filepath.Join(filepath.Dir(job.FilePath), baseName+outExt) + + duration, _ := m.encoder.ProbeDuration(job.FilePath) + + tempInPath := filepath.Join(m.TempDir, fmt.Sprintf("temp_in_%d_%s", job.ID, filepath.Base(job.FilePath))) + + var cmdErr error + var execCmd *exec.Cmd + + if job.MediaType == "video" { + if m.encoder.ShouldSkipVideo(job.FilePath, job.TargetResolution) { + log.Printf("Skipping video %d - already target codec/resolution", job.ID) + return db.AddJobReport(job, "skipped", originalSize, 0, 0) + } + + w, h, err := m.encoder.ProbeResolution(job.FilePath) + if err != nil { + return fmt.Errorf("failed to probe resolution: %w", err) + } + + log.Printf("Copying %s to %s before encoding...", job.FilePath, tempInPath) + if err := copyFile(job.FilePath, tempInPath); err != nil { + return fmt.Errorf("failed to copy source to temp: %w", err) + } + defer os.Remove(tempInPath) + + execCmd, err = m.encoder.BuildVideoCmd(tempInPath, tempOutPath, job.TargetResolution, job.FFmpegFlags, w, h) + if err != nil { + return err + } + } else { + if m.encoder.ShouldSkipAudio(job.FilePath) { + log.Printf("Skipping audio %d - already target codec/bitrate", job.ID) + return db.AddJobReport(job, "skipped", originalSize, 0, 0) + } + + log.Printf("Copying %s to %s before encoding...", job.FilePath, tempInPath) + if err := copyFile(job.FilePath, tempInPath); err != nil { + return fmt.Errorf("failed to copy source to temp: %w", err) + } + defer os.Remove(tempInPath) + + execCmd, cmdErr = m.encoder.BuildAudioCmd(tempInPath, tempOutPath, job.FFmpegFlags) + if cmdErr != nil { + return cmdErr + } + } + + stderr, err := execCmd.StderrPipe() + if err != nil { + return err + } + + if err := execCmd.Start(); err != nil { + return fmt.Errorf("failed to start ffmpeg: %w", err) + } + + scanner := bufio.NewScanner(stderr) + scanner.Split(bufio.ScanLines) + // Some ffmpeg progress output uses carriage returns instead of newlines + // We handle standard lines, to get a more robust carriage return split, we could implement a custom split function. + // For simplicity, ScanLines usually catches 'time=' lines well enough if they have \n. + // Let's use a custom split to handle '\r' as well. + scanner.Split(func(data []byte, atEOF bool) (advance int, token []byte, err error) { + if atEOF && len(data) == 0 { + return 0, nil, nil + } + if i := bytes.IndexByte(data, '\r'); i >= 0 { + return i + 1, data[0:i], nil + } + if i := bytes.IndexByte(data, '\n'); i >= 0 { + return i + 1, data[0:i], nil + } + if atEOF { + return len(data), data, nil + } + return 0, nil, nil + }) + + var lastErrLine string + go func() { + for scanner.Scan() { + line := scanner.Text() + lastErrLine = line + if duration > 0 { + prog := encoder.ParseProgress(line, duration) + if prog >= 0 { + m.NotifySSE("progress", map[string]interface{}{ + "id": job.ID, + "progress": fmt.Sprintf("%.1f", prog), + }) + } + } + } + }() + + if err := execCmd.Wait(); err != nil { + return fmt.Errorf("ffmpeg error: %v, last output: %s", err, lastErrLine) + } + + // Validate + if err := m.encoder.ValidateFile(tempOutPath); err != nil { + os.Remove(tempOutPath) + return fmt.Errorf("validation failed: %w", err) + } + + // Calculate sizes + outInfo, err := os.Stat(tempOutPath) + if err != nil { + return fmt.Errorf("failed to stat output: %w", err) + } + encodedSize := outInfo.Size() + sizeSaved := originalSize - encodedSize + + // Move file + log.Printf("Copying encoded file back to %s...", finalOutPath) + if err := os.Rename(tempOutPath, finalOutPath); err != nil { + // Fallback to copy if cross-device link error + if err := copyFile(tempOutPath, finalOutPath); err != nil { + return fmt.Errorf("failed to move output: %w", err) + } + os.Remove(tempOutPath) + } + + // Delete original if different extension + if finalOutPath != job.FilePath { + os.Remove(job.FilePath) + } + + processTime := time.Since(startTime).Seconds() + return db.AddJobReport(job, "success", encodedSize, sizeSaved, processTime) +} diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go new file mode 100644 index 0000000..03b28ac --- /dev/null +++ b/internal/watcher/watcher.go @@ -0,0 +1,222 @@ +package watcher + +import ( + "log" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/fsnotify/fsnotify" + "goencode/internal/db" + "goencode/internal/queue" +) + +type Manager struct { + watcher *fsnotify.Watcher + queueManager *queue.Manager + timers map[string]*time.Timer + timersMu sync.Mutex + processChan chan string + stopChan chan struct{} +} + +func NewManager(qm *queue.Manager) (*Manager, error) { + w, err := fsnotify.NewWatcher() + if err != nil { + return nil, err + } + + m := &Manager{ + watcher: w, + queueManager: qm, + timers: make(map[string]*time.Timer), + processChan: make(chan string, 10000), + stopChan: make(chan struct{}), + } + + for i := 0; i < 3; i++ { + go m.processWorker() + } + + return m, nil +} + +func (m *Manager) Start() { + m.Reload() + go m.watchLoop() +} + +func (m *Manager) Stop() { + close(m.stopChan) + m.watcher.Close() +} + +func (m *Manager) Reload() { + // Remove all existing watches + for _, path := range m.watcher.WatchList() { + m.watcher.Remove(path) + } + + folders, err := db.GetWatchFolders() + if err != nil { + log.Printf("Watcher failed to get folders: %v", err) + return + } + + for _, f := range folders { + if !f.Enabled { + continue + } + if err := os.MkdirAll(f.FolderPath, 0755); err != nil { + log.Printf("Failed to create watch folder %s: %v", f.FolderPath, err) + continue + } + log.Printf("Watching and scanning %s", f.FolderPath) + + // Walk the directory to add all subdirectories to watcher and scan existing files + filepath.Walk(f.FolderPath, func(path string, info os.FileInfo, err error) error { + if err != nil { + return nil + } + if info.IsDir() { + if err := m.watcher.Add(path); err != nil { + log.Printf("Failed to watch %s: %v", path, err) + } + } else { + // Process existing file asynchronously + go m.handleEvent(path) + } + return nil + }) + } +} + +func (m *Manager) watchLoop() { + for { + select { + case <-m.stopChan: + return + case event, ok := <-m.watcher.Events: + if !ok { + return + } + if event.Has(fsnotify.Write) || event.Has(fsnotify.Create) { + // Check if the created event is a directory + if event.Has(fsnotify.Create) { + info, err := os.Stat(event.Name) + if err == nil && info.IsDir() { + filepath.Walk(event.Name, func(path string, info os.FileInfo, err error) error { + if err != nil { + return nil + } + if info.IsDir() { + m.watcher.Add(path) + } else { + go m.handleEvent(path) + } + return nil + }) + } + } + m.handleEvent(event.Name) + } + case err, ok := <-m.watcher.Errors: + if !ok { + return + } + log.Printf("Watcher error: %v", err) + } + } +} + +func (m *Manager) processWorker() { + for { + select { + case <-m.stopChan: + return + case path := <-m.processChan: + m.processFile(path) + } + } +} + +func (m *Manager) handleEvent(filePath string) { + m.timersMu.Lock() + defer m.timersMu.Unlock() + + if t, exists := m.timers[filePath]; exists { + t.Stop() + } + + m.timers[filePath] = time.AfterFunc(5*time.Second, func() { + m.timersMu.Lock() + delete(m.timers, filePath) + m.timersMu.Unlock() + + select { + case m.processChan <- filePath: + default: + log.Printf("Process queue full, dropping %s", filePath) + } + }) +} + +func (m *Manager) processFile(filePath string) { + + info, err := os.Stat(filePath) + if err != nil || info.IsDir() { + return // File removed or is a directory + } + + // Make sure it's not a temp file + if filepath.Ext(filePath) == ".tmp" { + return + } + + alreadyInQueue, err := db.IsFileAlreadyProcessedOrQueued(filePath) + if err != nil { + log.Printf("Error checking DB for %s: %v", filePath, err) + return + } + if alreadyInQueue { + return + } + + // Find which watch folder it belongs to + folders, err := db.GetWatchFolders() + if err != nil { + return + } + + var match db.WatchFolder + found := false + for _, f := range folders { + if !f.Enabled { + continue + } + // Check if filePath is inside f.FolderPath + cleanPath := filepath.Clean(filePath) + folderPath := filepath.Clean(f.FolderPath) + if strings.HasPrefix(cleanPath, folderPath+string(os.PathSeparator)) || cleanPath == folderPath || filepath.Dir(cleanPath) == folderPath { + match = f + found = true + break + } + } + + if !found { + return + } + + err = db.AddJob(filePath, match.MediaType, 0, match.TargetResolution, match.CustomFFmpegFlags, info.Size()) + if err != nil { + log.Printf("Failed to add job for %s: %v", filePath, err) + return + } + + log.Printf("Added job for %s", filePath) + m.queueManager.NotifySSE("job_added", nil) + m.queueManager.Trigger() +} diff --git a/internal/web/api.go b/internal/web/api.go new file mode 100644 index 0000000..93e1b4e --- /dev/null +++ b/internal/web/api.go @@ -0,0 +1,103 @@ +package web + +import ( + "encoding/json" + "net/http" + "strconv" + "strings" + + "goencode/internal/db" +) + +func (s *Server) handleGetQueue(w http.ResponseWriter, r *http.Request) { + jobs, err := db.GetPendingJobs() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(jobs) +} + +func (s *Server) handleBumpJob(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + id, err := strconv.Atoi(strings.TrimPrefix(r.URL.Path, "/api/jobs/bump/")) + if err != nil { + http.Error(w, "Invalid ID", http.StatusBadRequest) + return + } + if err := db.BumpJobPriority(id); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + s.qm.NotifySSE("queue_updated", nil) + w.WriteHeader(http.StatusOK) +} + +func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost && r.Method != http.MethodDelete { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + id, err := strconv.Atoi(strings.TrimPrefix(r.URL.Path, "/api/jobs/cancel/")) + if err != nil { + http.Error(w, "Invalid ID", http.StatusBadRequest) + return + } + if err := db.DeleteJob(id); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + s.qm.NotifySSE("queue_updated", nil) + w.WriteHeader(http.StatusOK) +} + +func (s *Server) handleGetWatchFolders(w http.ResponseWriter, r *http.Request) { + folders, err := db.GetWatchFolders() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(folders) +} + +func (s *Server) handleAddWatchFolder(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + var f db.WatchFolder + if err := json.NewDecoder(r.Body).Decode(&f); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + f.Enabled = true + if err := db.AddWatchFolder(f); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + s.wm.Reload() + w.WriteHeader(http.StatusOK) +} + +func (s *Server) handleDeleteWatchFolder(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + id, err := strconv.Atoi(strings.TrimPrefix(r.URL.Path, "/api/folders/delete/")) + if err != nil { + http.Error(w, "Invalid ID", http.StatusBadRequest) + return + } + if err := db.DeleteWatchFolder(id); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + s.wm.Reload() + w.WriteHeader(http.StatusOK) +} diff --git a/internal/web/server.go b/internal/web/server.go new file mode 100644 index 0000000..d759701 --- /dev/null +++ b/internal/web/server.go @@ -0,0 +1,246 @@ +package web + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "html/template" + "log" + "net/http" + "path/filepath" + + "goencode/internal/config" + "goencode/internal/db" + "goencode/internal/logger" + "goencode/internal/queue" + "goencode/internal/watcher" + rootweb "goencode/web" +) + +type Server struct { + cfg *config.Config + qm *queue.Manager + wm *watcher.Manager + sse *SSEServer + mux *http.ServeMux + sessionToken string +} + +func NewServer(cfg *config.Config, qm *queue.Manager, wm *watcher.Manager, sse *SSEServer) *Server { + b := make([]byte, 32) + rand.Read(b) + token := hex.EncodeToString(b) + + s := &Server{ + cfg: cfg, + qm: qm, + wm: wm, + sse: sse, + mux: http.NewServeMux(), + sessionToken: token, + } + s.routes() + return s +} + +func (s *Server) Start() error { + addr := fmt.Sprintf("%s:%d", s.cfg.Server.ListenAddr, s.cfg.Server.Port) + + var handler http.Handler = s.mux + if s.cfg.Auth.Username != "" { + handler = s.authMiddleware(s.mux) + } + + return http.ListenAndServe(addr, handler) +} + +func (s *Server) authMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + importStrings := true // to ensure "strings" is imported if not already, wait, I need to check if strings is imported. + _ = importStrings + + // We'll use a simple slice of prefixes to bypass auth, but since there's only one, + // let's just use string slicing or import "strings" + + // To be safe without adding imports manually if I don't know the exact list: + if s.cfg.Auth.Username == "" || r.URL.Path == "/login" || (len(r.URL.Path) >= 8 && r.URL.Path[:8] == "/static/") { + next.ServeHTTP(w, r) + return + } + + cookie, err := r.Cookie("goencode_session") + if err != nil || cookie.Value != s.sessionToken { + http.Redirect(w, r, "/login", http.StatusFound) + return + } + + next.ServeHTTP(w, r) + }) +} + +func (s *Server) handleLogin(w http.ResponseWriter, r *http.Request) { + if r.Method == "GET" { + tmpl, err := template.ParseFS(rootweb.FS, "templates/login.html") + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + tmpl.Execute(w, nil) + return + } + + if r.Method == "POST" { + user := r.FormValue("username") + pass := r.FormValue("password") + + if user == s.cfg.Auth.Username && pass == s.cfg.Auth.Password { + http.SetCookie(w, &http.Cookie{ + Name: "goencode_session", + Value: s.sessionToken, + Path: "/", + HttpOnly: true, + }) + http.Redirect(w, r, "/", http.StatusFound) + return + } + + tmpl, _ := template.ParseFS(rootweb.FS, "templates/login.html") + tmpl.Execute(w, struct{ Error string }{"Invalid credentials"}) + } +} + +func (s *Server) handleLogout(w http.ResponseWriter, r *http.Request) { + http.SetCookie(w, &http.Cookie{ + Name: "goencode_session", + Value: "", + Path: "/", + MaxAge: -1, + HttpOnly: true, + }) + http.Redirect(w, r, "/login", http.StatusFound) +} + +func (s *Server) routes() { + s.mux.HandleFunc("/login", s.handleLogin) + s.mux.HandleFunc("/logout", s.handleLogout) + + s.mux.HandleFunc("/api/sse", s.sse.HandleSSE) + + s.mux.HandleFunc("/api/queue", s.handleGetQueue) + s.mux.HandleFunc("/api/jobs/bump/", s.handleBumpJob) + s.mux.HandleFunc("/api/jobs/cancel/", s.handleCancelJob) + + s.mux.HandleFunc("/api/folders", s.handleGetWatchFolders) + s.mux.HandleFunc("/api/folders/add", s.handleAddWatchFolder) + s.mux.HandleFunc("/api/folders/delete/", s.handleDeleteWatchFolder) + s.mux.HandleFunc("/api/logs", s.handleGetLogs) + + // Static files + s.mux.Handle("/static/", http.FileServer(http.FS(rootweb.FS))) + + // Pages + s.mux.HandleFunc("/", s.handleDashboard) + s.mux.HandleFunc("/folders", s.handlePage("folders.html")) + s.mux.HandleFunc("/history", s.handleHistory) + s.mux.HandleFunc("/logs", s.handlePage("logs.html")) + s.mux.HandleFunc("/settings", s.handlePage("settings.html")) +} + +func (s *Server) handleGetLogs(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + logs := logger.GetRecentLogs() + json.NewEncoder(w).Encode(logs) +} + +func formatBytes(bytes int64) string { + const unit = 1024 + if bytes < unit && bytes > -unit { + return fmt.Sprintf("%d B", bytes) + } + div, exp := int64(unit), 0 + n := bytes + if n < 0 { + n = -n + } + for n >= unit*unit && exp < len("KMGTPE")-1 { + div *= unit + exp++ + n /= unit + } + return fmt.Sprintf("%.1f %cB", float64(bytes)/float64(div), "KMGTPE"[exp]) +} + +func (s *Server) handleDashboard(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + + stats, _ := db.GetDashboardStats() + + // Create formatted stats + data := struct { + AuthEnabled bool + FilesEncoded int + QueueLength int + SavedSpace string + }{ + AuthEnabled: s.cfg.Auth.Username != "", + FilesEncoded: stats.FilesEncoded, + QueueLength: stats.QueueLength, + SavedSpace: formatBytes(stats.TotalSavedSpace), + } + + tmpl, err := template.New("layout").Funcs(template.FuncMap{ + "formatBytes": formatBytes, + }).ParseFS(rootweb.FS, "templates/layout.html", "templates/dashboard.html") + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + tmpl.ExecuteTemplate(w, "layout", data) +} + +func (s *Server) handlePage(tmplName string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + tmpl, err := template.ParseFS(rootweb.FS, "templates/layout.html", filepath.Join("templates", tmplName)) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + data := struct{ AuthEnabled bool }{AuthEnabled: s.cfg.Auth.Username != ""} + if err := tmpl.ExecuteTemplate(w, "layout", data); err != nil { + log.Printf("Template execution error: %v", err) + } + } +} + +func (s *Server) handleHistory(w http.ResponseWriter, r *http.Request) { + reports, total, err := db.GetJobReports(50, 0) // Basic pagination for now + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + data := struct { + AuthEnabled bool + Reports []db.JobReport + Total int + }{ + AuthEnabled: s.cfg.Auth.Username != "", + Reports: reports, + Total: total, + } + + tmpl, err := template.New("layout").Funcs(template.FuncMap{ + "formatBytes": formatBytes, + }).ParseFS(rootweb.FS, "templates/layout.html", "templates/history.html") + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if err := tmpl.ExecuteTemplate(w, "layout", data); err != nil { + log.Printf("Template execution error: %v", err) + } +} diff --git a/internal/web/sse.go b/internal/web/sse.go new file mode 100644 index 0000000..3652186 --- /dev/null +++ b/internal/web/sse.go @@ -0,0 +1,75 @@ +package web + +import ( + "encoding/json" + "fmt" + "net/http" + "sync" + "time" +) + +type SSEMessage struct { + Event string + Data interface{} +} + +type SSEServer struct { + clients map[chan SSEMessage]bool + mu sync.Mutex +} + +func NewSSEServer() *SSEServer { + return &SSEServer{ + clients: make(map[chan SSEMessage]bool), + } +} + +func (s *SSEServer) HandleSSE(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + + clientChan := make(chan SSEMessage) + s.mu.Lock() + s.clients[clientChan] = true + s.mu.Unlock() + + defer func() { + s.mu.Lock() + delete(s.clients, clientChan) + s.mu.Unlock() + close(clientChan) + }() + + notify := r.Context().Done() + + for { + select { + case <-notify: + return + case msg := <-clientChan: + dataBytes, err := json.Marshal(msg.Data) + if err != nil { + continue + } + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", msg.Event, string(dataBytes)) + w.(http.Flusher).Flush() + case <-time.After(30 * time.Second): + // Keepalive + fmt.Fprintf(w, "event: ping\ndata: {}\n\n") + w.(http.Flusher).Flush() + } + } +} + +func (s *SSEServer) Broadcast(event string, data interface{}) { + msg := SSEMessage{Event: event, Data: data} + s.mu.Lock() + defer s.mu.Unlock() + for clientChan := range s.clients { + select { + case clientChan <- msg: + default: + } + } +} diff --git a/web/fs.go b/web/fs.go new file mode 100644 index 0000000..ef8f6e7 --- /dev/null +++ b/web/fs.go @@ -0,0 +1,6 @@ +package web + +import "embed" + +//go:embed static/* templates/* +var FS embed.FS diff --git a/web/static/icon.png b/web/static/icon.png new file mode 100644 index 0000000..bb4789f Binary files /dev/null and b/web/static/icon.png differ diff --git a/web/static/style.css b/web/static/style.css new file mode 100644 index 0000000..ae419f9 --- /dev/null +++ b/web/static/style.css @@ -0,0 +1,263 @@ +:root { + --bg: #0a0a0a; + --surface: #141414; + --surface-hover: #1f1f1f; + --border: #2a2a2a; + --text-primary: #f0f0f0; + --text-secondary: #888888; + --accent: #21D198; + --accent-hover: #1cb582; + --success: #21D198; + --warning: #f59e0b; + --error: #ef4444; + --font: 'Inter', system-ui, -apple-system, sans-serif; +} + +* { + box-sizing: border-box; +} + +body { + margin: 0; + background-color: var(--bg); + color: var(--text-primary); + font-family: var(--font); + line-height: 1.6; +} + +a { + text-decoration: none; + color: inherit; +} + +/* Navbar */ +.navbar { + display: flex; + justify-content: space-between; + align-items: center; + padding: 1rem 2rem; + background: var(--surface); + border-bottom: 1px solid var(--border); +} + +.navbar-brand { + font-size: 1.5rem; + font-weight: 700; + color: var(--text-primary); +} + +.navbar-brand span { + color: var(--accent); +} + +.nav-links { + display: flex; + gap: 2rem; +} + +.nav-links a { + color: var(--text-secondary); + font-weight: 500; + transition: color 0.2s; +} + +.nav-links a:hover, .nav-links a.active { + color: var(--accent); +} + +/* Layout */ +.container { + padding: 2rem; + max-width: 1400px; + margin: 0 auto; +} + +h1, h2, h3 { + margin-top: 0; + font-weight: 500; +} + +/* Cards */ +.card { + background: var(--surface); + border: 1px solid var(--border); + border-radius: 8px; + padding: 1.5rem; + margin-bottom: 1.5rem; +} + +/* Tables */ +.table-container { + overflow-x: auto; +} + +table { + width: 100%; + border-collapse: collapse; + font-size: 0.875rem; +} + +th, td { + padding: 1rem; + text-align: left; + border-bottom: 1px solid var(--border); +} + +th { + color: var(--text-secondary); + font-weight: 500; + text-transform: uppercase; + letter-spacing: 0.05em; + font-size: 0.75rem; +} + +tbody tr { + transition: background 0.2s; +} + +tbody tr:hover { + background: rgba(255, 255, 255, 0.02); +} + +/* Buttons */ +.btn { + background: var(--accent); + color: #000; + border: 1px solid var(--accent); + padding: 0.5rem 1rem; + border-radius: 4px; + cursor: pointer; + font-family: inherit; + font-weight: 600; + font-size: 0.875rem; + transition: all 0.2s ease; +} + +.btn:hover { + background: var(--accent-hover); + border-color: var(--accent-hover); + box-shadow: 0 0 10px rgba(33, 209, 152, 0.3); +} + +.btn-sm { + padding: 0.25rem 0.75rem; + font-size: 0.75rem; +} + +.btn-danger { + background: transparent; + color: var(--error); + border: 1px solid var(--error); +} + +.btn-danger:hover { + background: rgba(239, 68, 68, 0.1); + border-color: var(--error); + box-shadow: none; +} + +/* Badges */ +.badge { + padding: 0.25rem 0.6rem; + border-radius: 4px; + font-size: 0.7rem; + font-weight: 600; + text-transform: uppercase; + border: 1px solid transparent; +} + +.badge.processing { background: rgba(59, 130, 246, 0.1); color: #60a5fa; border-color: rgba(59, 130, 246, 0.3); } +.badge.pending { background: rgba(245, 158, 11, 0.1); color: var(--warning); border-color: rgba(245, 158, 11, 0.3); } +.badge.success { background: rgba(33, 209, 152, 0.1); color: var(--success); border-color: rgba(33, 209, 152, 0.3); } +.badge.failed { background: rgba(239, 68, 68, 0.1); color: var(--error); border-color: rgba(239, 68, 68, 0.3); } +.badge.skipped { background: rgba(136, 136, 136, 0.1); color: var(--text-secondary); border-color: rgba(136, 136, 136, 0.3); } + +/* Progress bar */ +.progress-bar { + width: 100%; + height: 8px; + background: var(--surface-hover); + border-radius: 4px; + overflow: hidden; + margin-top: 0.5rem; +} + +.progress-fill { + height: 100%; + background: var(--accent); + width: 0%; + transition: width 0.3s ease; +} + +/* Forms */ +.form-group { + margin-bottom: 1rem; +} + +.form-group label { + display: block; + margin-bottom: 0.5rem; + color: var(--text-secondary); + font-size: 0.875rem; +} + +.form-control { + width: 100%; + padding: 0.75rem; + background: var(--bg); + border: 1px solid var(--border); + border-radius: 4px; + color: var(--text-primary); + font-family: inherit; +} + +.form-control:focus { + outline: none; + border-color: var(--accent); +} + +/* Responsive adjustments */ +@media (max-width: 768px) { + .navbar { + flex-direction: column; + align-items: flex-start; + padding: 1rem; + } + + .nav-links { + margin-top: 1rem; + flex-wrap: wrap; + gap: 1rem; + } + + .nav-links a[href="/logout"] { + margin-left: 0 !important; + } + + .container { + padding: 1rem; + } + + .card { + padding: 1rem; + max-width: 100%; + } + + .table-container { + max-width: 100%; + } + + table { + white-space: nowrap; + } + + th, td { + padding: 0.75rem 0.5rem; + } + + .header { + flex-direction: column; + align-items: flex-start !important; + gap: 1rem; + } +} diff --git a/web/templates/dashboard.html b/web/templates/dashboard.html new file mode 100644 index 0000000..afaa0d7 --- /dev/null +++ b/web/templates/dashboard.html @@ -0,0 +1,123 @@ +{{define "content"}} +
+
+

Files Encoded

+
{{.FilesEncoded}}
+
+
+

Space Saved

+
{{.SavedSpace}}
+
+
+

In Queue

+
{{.QueueLength}}
+
+
+ +
+

Active Job Queue

+
+ + + + + + + + + + + + + + +
IDFileTypePriorityStatusActions
+
+
+ + +{{end}} diff --git a/web/templates/folders.html b/web/templates/folders.html new file mode 100644 index 0000000..0c63a63 --- /dev/null +++ b/web/templates/folders.html @@ -0,0 +1,152 @@ +{{define "content"}} +
+
+

Configured Watch Folders

+
+ + + + + + + + + + + + + +
PathTypeResolutionFlagsActions
+
+
+ +
+

Add Watch Folder

+
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+ +
+
+
+ + +{{end}} \ No newline at end of file diff --git a/web/templates/history.html b/web/templates/history.html new file mode 100644 index 0000000..3732839 --- /dev/null +++ b/web/templates/history.html @@ -0,0 +1,80 @@ +{{define "content"}} +
+
+

History

+
+ + Total Records: {{.Total}} +
+
+
+ + + + + + + + + + + + + + + {{range .Reports}} + + + + + + + + + + + {{else}} + + + + {{end}} + +
FileTypeStatusOriginal SizeEncoded SizeSavedTimeDate
+
{{.FilePath}}
+ {{if .ErrorMessage}} +
{{.ErrorMessage}}
+ {{end}} + 		{{.MediaType}}{{.Status}}{{formatBytes .OriginalSize}}{{formatBytes .EncodedSize}}{{formatBytes .SizeSaved}}{{.ProcessingTime}}s{{.CreatedAt.Format "Jan 02, 15:04:05"}}
No job reports found.
+
+
+ + +{{end}} diff --git a/web/templates/layout.html b/web/templates/layout.html new file mode 100644 index 0000000..741bd16 --- /dev/null +++ b/web/templates/layout.html @@ -0,0 +1,52 @@ +{{define "layout"}} + + + + + + GoEncode + + + + + + +
+ {{template "content" .}} +
+ + + + +{{end}} diff --git a/web/templates/login.html b/web/templates/login.html new file mode 100644 index 0000000..c61474e --- /dev/null +++ b/web/templates/login.html @@ -0,0 +1,140 @@ + + + + + + GoEncode - Login + + + + + +
+
+ +

GoEncode

+
+ + {{if .Error}} +
{{.Error}}
+ {{end}} + +
+
+ + +
+
+ + +
+ +
+
+ + diff --git a/web/templates/logs.html b/web/templates/logs.html new file mode 100644 index 0000000..53a8394 --- /dev/null +++ b/web/templates/logs.html @@ -0,0 +1,60 @@ +{{define "content"}} +
+
+

System Logs

+

Live streaming output of the application logs.

+
+
+ +
+ 

+

+
+
+{{end}}
diff --git a/web/templates/settings.html b/web/templates/settings.html
new file mode 100644
index 0000000..2d0ce6f
--- /dev/null
+++ b/web/templates/settings.html
@@ -0,0 +1,21 @@
+{{define "content"}}
+

+    
Global Settings

+    

+        Database and core server configurations are managed via goencode.yaml.
+        Application preferences can be configured below.
+    

+
+    

+        

+            
+            
+        

+        

+            
+            
+        

+        
+    

+

+{{end}}
\ No newline at end of file