infoalloggi-monorepo/apps/backend/video.go

package main

import (
	"backend/config"
	"backend/queries"
	"backend/typesdefs"
	"backend/utils"
	"fmt"
	"image/jpeg"
	"io"
	"log"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/labstack/echo/v4"
	"github.com/nfnt/resize"
	"github.com/nickalie/go-webpbin"
	"github.com/schollz/progressbar/v3"
)

func AddToVideoProcessing(tmp *typesdefs.AnnuncioParsed, annuncio *typesdefs.AnnuncioXML, p *[]typesdefs.MediaToProcess) {

	miogestVideos := []string{}
	if annuncio.AMMedias != nil {
		for _, media := range annuncio.AMMedias.AMVideo {
			if media.Url != nil {
				if strings.Contains(*media.Url, "video.miogest") {
					miogestVideos = append(miogestVideos, *media.Url)
				}

			}
		}
	}

	for idx, url := range miogestVideos {
		*p = append(*p, typesdefs.MediaToProcess{
			Codice: tmp.Codice,
			Url:    url,
			Order:  idx,
		})
	}
}

func ProcessVideoPool(toProcess *[]typesdefs.MediaToProcess) ([]typesdefs.UploadedMedia, error) {
	if len(*toProcess) == 0 {
		log.Println("No new videos to process.")
		return []typesdefs.UploadedMedia{}, nil
	}
	var (
		ProcessedVideos []typesdefs.ProcessedMedia
		UploadedVideos  []typesdefs.UploadedMedia
		mu              sync.Mutex
		wg              sync.WaitGroup
	)
	totalToProcess := len(*toProcess)
	bar := progressbar.Default(int64(totalToProcess), "Processing videos")
	sem := make(chan struct{}, config.Cfg.Videos.ConcurrentLimit)

	// clear videos folder
	err := clearVideoFolder()
	if err != nil {
		log.Fatalf("Failed to initialize image processing: %v", err)
	}

	// process videos
	for _, video := range *toProcess {
		sem <- struct{}{} // acquire semaphore
		wg.Add(1)
		go func(video typesdefs.MediaToProcess) {
			defer wg.Done()
			defer func() { <-sem }() // release semaphore
			videoPath := filepath.Join(config.Cfg.Videos.Path, fmt.Sprintf("%s_%d%s", video.Codice, video.Order, filepath.Ext(video.Url)))
			err := downloadVideo(video.Url, videoPath)
			if err != nil {
				log.Printf("Failed to download video %s: %v", video.Url, err)
				return // Exit goroutine on download failure
			}
			// Transcode and generate thumbnail
			mp4Path, thumbnailPath, err := VideoConversion(videoPath)
			if err != nil {
				log.Printf("Failed to convert video %s: %v", videoPath, err)
				return
			}
			mu.Lock()
			ProcessedVideos = append(ProcessedVideos, typesdefs.ProcessedMedia{
				Codice:       video.Codice,
				Order:        video.Order,
				Path:         mp4Path,
				ThumnailPath: thumbnailPath,
				OGUrl:        video.Url,
			})
			bar.Add(1)
			mu.Unlock()
		}(video)
	}
	wg.Wait()
	bar.Finish()
	fmt.Println()

	totalToUpload := len(ProcessedVideos)
	uploadBar := progressbar.Default(int64(totalToUpload), "Uploading videos")
	uploadSem := make(chan struct{}, config.Cfg.Videos.ConcurrentLimit)
	var uploadWg sync.WaitGroup
	for _, video := range ProcessedVideos {
		uploadSem <- struct{}{} // acquire semaphore
		uploadWg.Add(1)
		go func(video typesdefs.ProcessedMedia) {
			defer uploadWg.Done()
			defer func() { <-uploadSem }() // release semaphore

			videoId, err := UploadFile(video.Path, "video/mp4", "videos")
			if err != nil {
				log.Printf("failed to upload video: %v", err)
				return
			}
			thumbId, err := UploadFile(video.ThumnailPath, "image/webp", "videos")
			if err != nil {
				log.Printf("failed to upload thumbnail: %v", err)
				return
			}
			mu.Lock()
			UploadedVideos = append(UploadedVideos, typesdefs.UploadedMedia{
				Codice:  video.Codice,
				Order:   video.Order,
				MediaId: videoId,
				ThumbId: thumbId,
				OGUrl:   video.OGUrl,
			})
			uploadBar.Add(1)
			mu.Unlock()
		}(video)
	}
	uploadWg.Wait()
	uploadBar.Finish()
	fmt.Println()

	return UploadedVideos, nil
}

func clearVideoFolder() error {
	err := utils.SafeRemoveAll(config.Cfg.Videos.Path)
	if err != nil {
		return err
	}
	err = os.Mkdir(config.Cfg.Videos.Path, 0755)
	if err != nil {
		return err
	}
	return nil
}

func downloadVideo(url, filename string) error {
	client := &http.Client{
		Timeout: 5 * time.Minute, // Adjust based on expected video sizes
	}

	response, err := client.Get(url)
	if err != nil {
		return fmt.Errorf("failed to download video: %w", err)
	}
	defer response.Body.Close()

	// Check HTTP status
	if response.StatusCode != http.StatusOK {
		return fmt.Errorf("failed to download video: HTTP %d", response.StatusCode)
	}

	// Optional: Validate Content-Type
	contentType := response.Header.Get("Content-Type")
	if !strings.HasPrefix(contentType, "video/") && !strings.HasPrefix(contentType, "application/") {
		log.Printf("Warning: unexpected content type %s for %s", contentType, url)
	}

	file, err := os.Create(filename)
	if err != nil {
		return fmt.Errorf("failed to create video file: %w", err)
	}
	defer file.Close()

	_, err = io.Copy(file, response.Body)
	if err != nil {
		return fmt.Errorf("failed to save video file: %w", err)
	}

	return nil
}

// VideoConversion transcodes video and generates thumbnail
func VideoConversion(input string) (string, string, error) {
	outputdir := filepath.Dir(input)
	basename := filepath.Base(input)
	ext := filepath.Ext(basename)
	nameWithoutExt := basename[0 : len(basename)-len(ext)]

	// Final destinations on the (potentially overlayfs) volume
	videoOutput := filepath.Join(outputdir, nameWithoutExt+".mp4")
	thumbOutput := filepath.Join(outputdir, "thumbnail-"+nameWithoutExt+".webp")

	// Use /tmp for all ffmpeg work: /tmp is tmpfs and supports random-access
	// seeks that overlayfs does not, which is required for MP4 muxing and
	// for the faststart moov-atom relocation.
	tmpVideo := filepath.Join(os.TempDir(), nameWithoutExt+"_converted.mp4")
	tmpThumb := filepath.Join(os.TempDir(), "thumbnail-"+nameWithoutExt+".webp")
	defer os.Remove(tmpVideo)
	defer os.Remove(tmpThumb)

	// 1. Transcode video to H.264/AAC MP4 (web-optimized) into /tmp
	err := transcodeVideo(input, tmpVideo)
	if err != nil {
		return "", "", fmt.Errorf("failed to transcode video: %w", err)
	}

	// 2. Extract thumbnail from original video into /tmp
	err = extractThumbnail(input, tmpThumb)
	if err != nil {
		return "", "", fmt.Errorf("failed to extract thumbnail: %w", err)
	}

	// 3. Move completed files from /tmp to the videos volume
	os.Remove(input)
	if err = copyFile(tmpVideo, videoOutput); err != nil {
		return "", "", fmt.Errorf("failed to move converted video: %w", err)
	}
	if err = copyFile(tmpThumb, thumbOutput); err != nil {
		return "", "", fmt.Errorf("failed to move thumbnail: %w", err)
	}

	return videoOutput, thumbOutput, nil
}

// copyFile copies src to dst, creating dst if needed.
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()
	if _, err = io.Copy(out, in); err != nil {
		return err
	}
	return out.Sync()
}

// transcodeVideo converts video to web-optimized MP4.
// output must be on a filesystem that supports random-access seeks (e.g. /tmp),
// because the MP4 muxer and -movflags +faststart both require re-opening the
// output file to relocate the moov atom — this fails on Docker overlayfs.
func transcodeVideo(input, output string) error {
	cmd := exec.Command("ffmpeg",
		"-i", input,
		"-c:v", "libx264",
		"-preset", "fast",
		"-crf", "23",
		"-c:a", "aac",
		"-b:a", "128k",
		"-movflags", "+faststart",
		"-vf", "scale=min(1280\\,iw):-2",
		"-y",
		output,
	)

	output_bytes, err := cmd.CombinedOutput()
	if err != nil {
		return fmt.Errorf("ffmpeg transcode failed: %w, output: %s", err, string(output_bytes))
	}

	return nil
}

// extractThumbnail generates a WebP thumbnail from video
func extractThumbnail(input, output string) error {
	// Extract frame at 1 second as JPEG to temp file
	tempJpeg := output + ".tmp.jpg"
	defer os.Remove(tempJpeg)

	cmd := exec.Command("ffmpeg",
		"-i", input,
		"-ss", "00:00:01", // seek to 1 second
		"-vframes", "1", // extract 1 frame
		"-q:v", "2", // JPEG quality (2-31, lower = better)
		"-y",
		tempJpeg,
	)

	output_bytes, err := cmd.CombinedOutput()
	if err != nil {
		return fmt.Errorf("ffmpeg thumbnail extraction failed: %w, output: %s", err, string(output_bytes))
	}

	// Convert JPEG to WebP thumbnail
	f, err := os.Open(tempJpeg)
	if err != nil {
		return fmt.Errorf("failed to open temp jpeg: %w", err)
	}
	defer f.Close()

	img, err := jpeg.Decode(f)
	if err != nil {
		return fmt.Errorf("failed to decode jpeg: %w", err)
	}

	// Resize to thumbnail (320px width)
	resizedImg := resize.Resize(320, 0, img, resize.Lanczos3)

	out, err := os.Create(output)
	if err != nil {
		return fmt.Errorf("failed to create thumbnail: %w", err)
	}
	defer out.Close()

	err = webpbin.Encode(out, resizedImg)
	if err != nil {
		return fmt.Errorf("failed to encode webp: %w", err)
	}

	return nil
}

func FallbackVideo(c echo.Context) error {
	// Serve a fallback image if the requested image is not found
	fallbackPath := "fallback.webp" // Path to your fallback image
	if _, err := os.Stat(fallbackPath); os.IsNotExist(err) {
		return c.String(http.StatusNotFound, "Fallback image not found")
	}
	return c.File(fallbackPath)
}

type videRef struct {
	Og_url string
	Video  string
	Thumb  string
}

func CodiciToProcessVideos(annunci *typesdefs.AnnunciXML) ([]string, error) {
	currentBucketState, err := ListBucket("videos")
	if err != nil {
		return nil, fmt.Errorf("failed to list bucket: %w", err)
	}

	//make a list of bucket ids
	bucketIdsMap := make(map[string]bool)
	for _, item := range currentBucketState {
		bucketIdsMap[item.ID] = true
	}

	db_videos, err := queries.GetVideosFromDB()
	if err != nil {
		return nil, err
	}

	dbVideosMap := make(map[string][]videRef) // Convert slice to map for faster lookup
	for _, ref := range db_videos {
		dbVideosMap[ref.Codice] = append(dbVideosMap[ref.Codice], videRef{
			Og_url: ref.Og_url,
			Video:  ref.Video,
			Thumb:  ref.Thumb,
		})
	}

	var codiciToProcess []string
	for _, annuncio := range annunci.Annuncio {

		if annuncio.Codice != nil && annuncio.AMMedias != nil {
			mediaUrls := []string{}
			for _, media := range annuncio.AMMedias.AMVideo {
				if media.Url != nil {
					if strings.Contains(*media.Url, "video.miogest") {
						mediaUrls = append(mediaUrls, *media.Url)
					}
				}
			}
			if len(mediaUrls) == 0 {
				continue
			}

			refVideos, existsInDb := dbVideosMap[*annuncio.Codice]
			if !existsInDb {
				codiciToProcess = append(codiciToProcess, *annuncio.Codice)
				continue
			}
			needProcess := false

			for _, media := range annuncio.AMMedias.AMVideo {
				if media.Url == nil {
					continue
				}
				if !strings.Contains(*media.Url, "video.miogest") {
					continue
				}

				found := false
				for _, ref := range refVideos {
					if *media.Url == ref.Og_url {
						if bucketIdsMap[ref.Video] && bucketIdsMap[ref.Thumb] {
							found = true
							break
						}
					}
				}
				if !found {
					needProcess = true
					break
				}

			}
			if needProcess {
				codiciToProcess = append(codiciToProcess, *annuncio.Codice)
			}
		}
	}

	return codiciToProcess, nil

}

func ForceClearVideos(codici []string) error {
	var idsToDelete []string
	for _, codice := range codici {
		refs, err := queries.GetCodiceVideosFromDB(codice)
		if err != nil {
			return fmt.Errorf("failed to get videos by codice %s: %w", codice, err)
		}
		for _, ref := range refs {
			idsToDelete = append(idsToDelete, ref.Video)
			idsToDelete = append(idsToDelete, ref.Thumb)
		}

	}
	err := DeleteIdsFromStorage(idsToDelete)
	if err != nil {
		return fmt.Errorf("failed to delete videos from storage: %w", err)
	}
	for _, codice := range codici {
		err := queries.ClearVideosRefsByCodice(codice)
		if err != nil {
			return fmt.Errorf("failed to clear videos by codice %s: %w", codice, err)
		}
	}
	return nil
}

// ReconcileVideos removes stranded videos (in storage but not in DB)
func ReconcileVideos() error {
	currentBucketState, err := ListBucket("videos")
	if err != nil {
		return fmt.Errorf("failed to list bucket: %w", err)
	}
	//make a list of bucket ids
	bucketIdsMap := make(map[string]bool)
	for _, item := range currentBucketState {
		bucketIdsMap[item.ID] = true
	}
	db_videos, err := queries.GetVideosFromDB()
	if err != nil {
		return fmt.Errorf("failed to get videos from DB: %w", err)
	}
	var strandedIds []string
	for _, ref := range db_videos {
		if !bucketIdsMap[ref.Video] {
			strandedIds = append(strandedIds, ref.Video)
		}
		if !bucketIdsMap[ref.Thumb] {
			strandedIds = append(strandedIds, ref.Thumb)
		}
	}
	if len(strandedIds) > 0 {
		err := DeleteIdsFromStorage(strandedIds)
		if err != nil {
			return fmt.Errorf("failed to delete stranded videos from storage: %w", err)
		}
	}
	return nil
}