package photos

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"hash/fnv"
	"io"
	"log/slog"
	"os"
	"sync"
	"time"

	"galo/internal/auth"
	"galo/internal/video"
)

const maxVariantAttempts = 5

// videoJobLease is the processing lease for a video job. It is far longer than
// the image lease because a transcode can run for minutes, and must outlast the
// worst-case processing so a job is never claimed twice concurrently.
const videoJobLease = 30 * time.Minute

// videoJobTimeout bounds a single ffmpeg/ffprobe invocation.
const videoJobTimeout = 20 * time.Minute

// Worker is the background reconciler. It drives the variant generation queue
// and the object garbage collector entirely from the state recorded in the
// metadata layer, so it resumes correctly after a restart.
type Worker struct {
	svc         *Service
	interval    time.Duration
	batch       int
	pendingTTL  time.Duration
	lease       time.Duration
	concurrency int

	tools      *video.Toolchain
	videoConc  int
	videoLease time.Duration
	tempDir    string

	// wake будит воркер немедленно, чтобы свежие правки не ждали тика.
	wake chan struct{}
}

// NewWorker builds a worker. Parameters are clamped to sane defaults.
func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time.Duration, concurrency int) *Worker {
	if interval <= 0 {
		interval = 5 * time.Second
	}
	if batch <= 0 {
		batch = 10
	}
	if pendingTTL <= 0 {
		pendingTTL = 10 * time.Minute
	}
	if concurrency <= 0 {
		concurrency = 4
	}
	return &Worker{
		svc:         svc,
		interval:    interval,
		batch:       batch,
		pendingTTL:  pendingTTL,
		lease:       interval * 4,
		concurrency: concurrency,
		videoConc:   1,
		videoLease:  videoJobLease,
		wake:        make(chan struct{}, 1),
	}
}

// Wake asks the worker to process the queues right away instead of waiting for
// the next tick. It is safe to call from any goroutine.
func (w *Worker) Wake() {
	if w.wake == nil {
		return
	}
	select {
	case w.wake <- struct{}{}:
	default:
	}
}

// SetVideo enables video processing with the given toolchain and a separate,
// small transcode concurrency. A nil or unavailable toolchain leaves video jobs
// untouched.
func (w *Worker) SetVideo(tools *video.Toolchain, concurrency int) {
	w.tools = tools
	if concurrency > 0 {
		w.videoConc = concurrency
	}
	w.videoLease = videoJobLease
	w.tempDir = tools.TempDir()
}

// Run loops until the context is cancelled.
func (w *Worker) Run(ctx context.Context) {
	slog.Info("background worker started", "interval", w.interval)
	defer slog.Info("background worker stopped")
	ticker := time.NewTicker(w.interval)
	defer ticker.Stop()
	for {
		w.tick(ctx)
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		case <-w.wake:
		}
	}
}

// ProcessOnce runs a single round of all queues. It is used by tests and manual
// reconciliation and does not block.
func (w *Worker) ProcessOnce(ctx context.Context) {
	w.tick(ctx)
}

func (w *Worker) tick(ctx context.Context) {
	w.processVariantJobs(ctx)
	w.processVideoJobs(ctx)
	w.processEditJobs(ctx)
	w.collectGarbage(ctx)
	w.reapStalePending(ctx)
}

func (w *Worker) processVariantJobs(ctx context.Context) {
	ids, err := w.svc.store.ClaimVariantJobs(ctx, KindImage, w.batch, w.lease)
	if err != nil {
		slog.Error("claim variant jobs", "error", err)
		return
	}
	w.forEachConcurrent(ids, func(id string) {
		w.generateVariants(ctx, id)
	})
}

// processVideoJobs claims and processes video jobs on a separate, small pool
// with a long lease so transcoding does not starve image variant generation.
func (w *Worker) processVideoJobs(ctx context.Context) {
	if w.tools == nil || !w.tools.Available() {
		return
	}
	ids, err := w.svc.store.ClaimVariantJobs(ctx, KindVideo, w.batch, w.videoLease)
	if err != nil {
		slog.Error("claim video jobs", "error", err)
		return
	}
	if len(ids) == 0 {
		return
	}
	pool := &Worker{concurrency: w.videoConc}
	pool.forEachConcurrent(ids, func(id string) {
		w.generateVideo(ctx, id)
	})
}

// processEditJobs claims and processes the pending renders of the edited
// versions of photos.
func (w *Worker) processEditJobs(ctx context.Context) {
	ids, err := w.svc.store.ClaimEditJobs(ctx, w.batch, w.lease)
	if err != nil {
		slog.Error("claim edit jobs", "error", err)
		return
	}
	w.forEachConcurrent(ids, func(id string) {
		w.generateEdit(ctx, id)
	})
}

// generateEdit renders the edited version of a photo from its original and
// the parameters stored in the metadata. The variants are bound next to the
// base ones, so the client can show the edited photo without touching the
// original object.
func (w *Worker) generateEdit(ctx context.Context, photoID string) {
	p, err := w.svc.store.GetByID(ctx, photoID, "")
	if err != nil {
		_ = w.svc.store.CompleteEditJob(ctx, photoID)
		return
	}
	params := EditParamsOf(p)
	if params.IsEmpty() {
		// The photo was reset while the job was queued.
		_ = w.svc.store.CompleteEditJob(ctx, photoID)
		return
	}
	orig, err := w.svc.store.OriginalObject(ctx, photoID)
	if err != nil {
		_ = w.svc.store.CompleteEditJob(ctx, photoID)
		return
	}
	if orig.State != ObjectReady {
		w.failEdit(ctx, photoID, "original object is not ready")
		return
	}
	data, err := w.svc.storage.Get(ctx, orig.StorageKey)
	if err != nil {
		w.failEdit(ctx, photoID, "fetch original: "+err.Error())
		return
	}
	edited, err := decodeAndAdjust(data, params, editSeed(photoID))
	if err != nil {
		w.failEdit(ctx, photoID, "decode original: "+err.Error())
		return
	}
	thumb, medium, preview, err := encodeEditedVariants(edited)
	if err != nil {
		w.failEdit(ctx, photoID, "build edited variants: "+err.Error())
		return
	}
	// The edit belongs to this photo only, so the variants are not shared
	// with the copies of the same original.
	owner := []string{photoID}
	if err := w.createVariant(ctx, owner, RoleEditThumb, thumb); err != nil {
		w.failEdit(ctx, photoID, "store edit thumb: "+err.Error())
		return
	}
	if err := w.createVariant(ctx, owner, RoleEditMedium, medium); err != nil {
		w.failEdit(ctx, photoID, "store edit medium: "+err.Error())
		return
	}
	if err := w.createVariant(ctx, owner, RoleEditPreview, preview); err != nil {
		w.failEdit(ctx, photoID, "store edit preview: "+err.Error())
		return
	}
	// The visible variants are ready: notify right away, so the grid and the
	// viewer show the edit. Полноразмерная версия готовится следом: пока она
	// не готова, браузер рисует видимую область сам.
	w.notify(ctx, photoID)
	if full, err := encodeEdited(edited); err == nil && len(full) > 0 {
		if err := w.createVariant(ctx, owner, RoleEdit, full); err != nil {
			w.failEdit(ctx, photoID, "store edit: "+err.Error())
			return
		}
	}
	_ = w.svc.store.CompleteEditJob(ctx, photoID)
	w.notify(ctx, photoID)
}

// editSeed is the deterministic seed of the grain render: the FNV-1a hash of the
// photo id. It depends on the photo and not on the parameters, so a repeated
// render of the same edit produces the same noise pattern.
func editSeed(photoID string) uint64 {
	h := fnv.New64a()
	_, _ = h.Write([]byte(photoID))
	return h.Sum64()
}

func (w *Worker) failEdit(ctx context.Context, photoID, msg string) {
	slog.Warn("edit job failed", "photo", photoID, "error", msg)
	_ = w.svc.store.FailEditJob(ctx, photoID, msg, maxVariantAttempts)
}

// forEachConcurrent runs fn over ids with at most w.concurrency functions in
// flight. Generation is CPU-bound (decode + resize + encode), so parallelism
// scales with cores; a concurrency of 1 stays strictly sequential.
func (w *Worker) forEachConcurrent(ids []string, fn func(string)) {
	if len(ids) == 0 {
		return
	}
	if w.concurrency <= 1 {
		for _, id := range ids {
			fn(id)
		}
		return
	}
	sem := make(chan struct{}, w.concurrency)
	var wg sync.WaitGroup
	for _, id := range ids {
		wg.Add(1)
		sem <- struct{}{}
		go func(id string) {
			defer wg.Done()
			defer func() { <-sem }()
			fn(id)
		}(id)
	}
	wg.Wait()
}

func (w *Worker) generateVariants(ctx context.Context, photoID string) {
	orig, err := w.svc.store.OriginalObject(ctx, photoID)
	if err != nil {
		// The photo or its original binding is gone; nothing to do.
		_ = w.svc.store.CompleteVariantJob(ctx, photoID)
		return
	}
	if orig.State != ObjectReady {
		w.fail(ctx, photoID, "original object is not ready")
		return
	}
	data, err := w.svc.storage.Get(ctx, orig.StorageKey)
	if err != nil {
		w.fail(ctx, photoID, "fetch original: "+err.Error())
		return
	}
	thumb, medium, preview, err := buildVariants(data)
	if err != nil {
		w.fail(ctx, photoID, "build variants: "+err.Error())
		return
	}
	siblings, err := w.svc.store.PhotosByOriginalObject(ctx, orig.ID)
	if err != nil || len(siblings) == 0 {
		siblings = []string{photoID}
	}
	if err := w.createVariant(ctx, siblings, RoleThumb, thumb); err != nil {
		w.fail(ctx, photoID, "store thumb: "+err.Error())
		return
	}
	if err := w.createVariant(ctx, siblings, RoleMedium, medium); err != nil {
		w.fail(ctx, photoID, "store medium: "+err.Error())
		return
	}
	if err := w.createVariant(ctx, siblings, RolePreview, preview); err != nil {
		w.fail(ctx, photoID, "store preview: "+err.Error())
		return
	}
	_ = w.svc.store.CompleteVariantJob(ctx, photoID)
	for _, pid := range siblings {
		w.notify(ctx, pid)
	}
}

// createVariant stores one derived image object and binds it to every photo
// sharing the original, so copies reuse the same physical variant.
func (w *Worker) createVariant(ctx context.Context, photoIDs []string, role string, data []byte) error {
	return w.createObject(ctx, photoIDs, role, ".webp", "image/webp", int64(len(data)), checksum(data), bytes.NewReader(data))
}

// createObject stores one derived object and binds it to every photo sharing
// the original.
func (w *Worker) createObject(ctx context.Context, photoIDs []string, role, ext, mimeType string, size int64, sum string, body io.Reader) error {
	objID := auth.NewID()
	key := objectKey(objID, role, ext)
	obj := &Object{
		ID:         objID,
		StorageKey: key,
		MimeType:   mimeType,
		Size:       size,
		Checksum:   sum,
	}
	if err := w.svc.store.CreateObject(ctx, obj); err != nil {
		return err
	}
	if err := w.svc.storage.Put(ctx, key, body, size, mimeType); err != nil {
		_ = w.svc.store.DeleteObjectRow(ctx, objID)
		return err
	}
	if err := w.svc.store.ActivateObject(ctx, objID); err != nil {
		return err
	}
	bound := false
	for _, pid := range photoIDs {
		if err := w.svc.store.BindObject(ctx, pid, objID, role); err != nil {
			if errors.Is(err, ErrObjectNotReady) || errors.Is(err, ErrObjectNotFound) {
				continue
			}
			return err
		}
		bound = true
	}
	if !bound {
		// No photo could take the variant; let the collector reclaim it.
		_ = w.svc.store.MarkObjectDeleting(ctx, objID)
	}
	return nil
}

// generateVideo prepares a video job: probe, poster frame (reusing the image
// variant pipeline) and a web-friendly playback variant, then records the
// probed facts and notifies subscribers.
func (w *Worker) generateVideo(ctx context.Context, photoID string) {
	orig, err := w.svc.store.OriginalObject(ctx, photoID)
	if err != nil {
		_ = w.svc.store.CompleteVariantJob(ctx, photoID)
		return
	}
	if orig.State != ObjectReady {
		w.fail(ctx, photoID, "original object is not ready")
		return
	}
	inPath, cleanup, err := w.spoolOriginal(ctx, orig.StorageKey)
	if err != nil {
		w.fail(ctx, photoID, "spool original: "+err.Error())
		return
	}
	defer cleanup()

	info, err := video.Probe(ctx, w.tools.FFprobePath(), inPath)
	if err != nil {
		w.fail(ctx, photoID, "probe: "+err.Error())
		return
	}

	siblings, err := w.svc.store.PhotosByOriginalObject(ctx, orig.ID)
	if err != nil || len(siblings) == 0 {
		siblings = []string{photoID}
	}

	if err := w.storePoster(ctx, inPath, siblings); err != nil {
		w.fail(ctx, photoID, "poster: "+err.Error())
		return
	}
	if err := w.storePlayback(ctx, inPath, info, siblings); err != nil {
		w.fail(ctx, photoID, "playback: "+err.Error())
		return
	}
	if err := w.svc.store.SetMediaMeta(ctx, photoID, orig.ID, info.DurationMS, info.Width, info.Height); err != nil {
		w.fail(ctx, photoID, "save media meta: "+err.Error())
		return
	}

	_ = w.svc.store.CompleteVariantJob(ctx, photoID)
	for _, pid := range siblings {
		w.notify(ctx, pid)
	}
}

// spoolOriginal streams the original blob into a temporary file so ffmpeg reads
// a seekable input. The returned cleanup removes the file.
func (w *Worker) spoolOriginal(ctx context.Context, key string) (string, func(), error) {
	src, err := w.svc.storage.Open(ctx, key)
	if err != nil {
		return "", nil, err
	}
	defer src.Close()
	f, err := os.CreateTemp(w.tempDir, "video-in-*")
	if err != nil {
		return "", nil, err
	}
	name := f.Name()
	cleanup := func() { _ = os.Remove(name) }
	if _, err := io.Copy(f, src); err != nil {
		_ = f.Close()
		cleanup()
		return "", nil, err
	}
	if err := f.Close(); err != nil {
		cleanup()
		return "", nil, err
	}
	return name, cleanup, nil
}

// storePoster extracts a frame, retrying at the start for videos whose first
// second is unavailable, then generates the image variants from it.
func (w *Worker) storePoster(ctx context.Context, inPath string, siblings []string) error {
	pf, err := os.CreateTemp(w.tempDir, "video-poster-*.jpg")
	if err != nil {
		return err
	}
	posterPath := pf.Name()
	_ = pf.Close()
	defer func() { _ = os.Remove(posterPath) }()

	// Seek to 1s for a representative frame, falling back to the first frame
	// when that yields no image (for example a clip shorter than one second).
	if err := w.runPoster(ctx, inPath, posterPath, 1); err != nil {
		if err := w.runPoster(ctx, inPath, posterPath, 0); err != nil {
			return err
		}
	}
	poster, err := os.ReadFile(posterPath)
	if err != nil {
		return err
	}
	if len(poster) == 0 {
		return errors.New("poster frame is empty")
	}
	thumb, medium, preview, err := buildVariants(poster)
	if err != nil {
		return err
	}
	if err := w.createVariant(ctx, siblings, RoleThumb, thumb); err != nil {
		return err
	}
	if err := w.createVariant(ctx, siblings, RoleMedium, medium); err != nil {
		return err
	}
	return w.createVariant(ctx, siblings, RolePreview, preview)
}

// runPoster extracts a poster frame at the given second and reports an error
// when ffmpeg fails or produces no image.
func (w *Worker) runPoster(ctx context.Context, inPath, outPath string, atSecond int) error {
	if err := w.runFFmpeg(ctx, video.PosterArgs(inPath, outPath, atSecond)); err != nil {
		return err
	}
	info, err := os.Stat(outPath)
	if err != nil {
		return err
	}
	if info.Size() == 0 {
		return errors.New("poster frame is empty")
	}
	return nil
}

// storePlayback produces the web-friendly mp4 and binds it as the playback
// role. Web-compatible sources are remuxed; everything else is transcoded.
func (w *Worker) storePlayback(ctx context.Context, inPath string, info *video.Info, siblings []string) error {
	of, err := os.CreateTemp(w.tempDir, "video-out-*.mp4")
	if err != nil {
		return err
	}
	outPath := of.Name()
	_ = of.Close()
	defer func() { _ = os.Remove(outPath) }()

	remuxable := info.CanRemux(video.MaxLongSide)
	if remuxable {
		if err := w.runFFmpeg(ctx, video.RemuxArgs(inPath, outPath)); err != nil {
			// Some remux-eligible files still fail; fall back to a transcode.
			if err2 := w.runFFmpeg(ctx, video.TranscodeArgs(inPath, outPath, video.MaxLongSide)); err2 != nil {
				return err
			}
		}
	} else if err := w.runFFmpeg(ctx, video.TranscodeArgs(inPath, outPath, video.MaxLongSide)); err != nil {
		return err
	}

	f, err := os.Open(outPath)
	if err != nil {
		return err
	}
	defer f.Close()
	stat, err := f.Stat()
	if err != nil {
		return err
	}
	sum, err := hashReader(f)
	if err != nil {
		return err
	}
	if _, err := f.Seek(0, io.SeekStart); err != nil {
		return err
	}
	return w.createObject(ctx, siblings, RolePlayback, ".mp4", "video/mp4", stat.Size(), sum, f)
}

// runFFmpeg runs ffmpeg with a bounded timeout.
func (w *Worker) runFFmpeg(ctx context.Context, args []string) error {
	tctx, cancel := context.WithTimeout(ctx, videoJobTimeout)
	defer cancel()
	return video.Run(tctx, w.tools.FFmpegPath(), args...)
}

func hashReader(r io.Reader) (string, error) {
	h := sha256.New()
	if _, err := io.Copy(h, r); err != nil {
		return "", err
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

func (w *Worker) notify(ctx context.Context, photoID string) {
	p, err := w.svc.store.GetByID(ctx, photoID, "")
	if err != nil {
		return
	}
	if err := w.svc.attachURLs(ctx, p); err != nil {
		return
	}
	// Variant readiness updates an existing photo; it is not a new photo in the
	// album, so clients update it in place instead of inserting at the top.
	w.svc.hub.Publish(p.AlbumID, "photo.updated", p)
	// The album object only carries a cover link once its variants exist, so the
	// cover's readiness is what actually refreshes album cards.
	w.svc.albums.NotifyCoverReady(ctx, p.AlbumID, p.ID)
}

func (w *Worker) fail(ctx context.Context, photoID, msg string) {
	slog.Warn("variant job failed", "photo", photoID, "error", msg)
	_ = w.svc.store.FailVariantJob(ctx, photoID, msg, maxVariantAttempts)
}

func (w *Worker) collectGarbage(ctx context.Context) {
	objs, err := w.svc.store.ClaimGarbage(ctx, w.batch)
	if err != nil {
		slog.Error("claim garbage", "error", err)
		return
	}
	for _, o := range objs {
		w.deleteObject(ctx, o)
	}
}

func (w *Worker) reapStalePending(ctx context.Context) {
	objs, err := w.svc.store.ClaimStalePending(ctx, w.pendingTTL, w.batch)
	if err != nil {
		slog.Error("claim stale pending", "error", err)
		return
	}
	for _, o := range objs {
		w.deleteObject(ctx, o)
	}
}

// deleteObject removes an object's blob then its row. If a binding appeared
// meanwhile the object is returned to ready; if the storage delete fails the
// row stays deleting and is retried on the next pass.
func (w *Worker) deleteObject(ctx context.Context, o Object) {
	if n, err := w.svc.store.ObjectRefCount(ctx, o.ID); err == nil && n > 0 {
		_ = w.svc.store.SetObjectState(ctx, o.ID, ObjectReady)
		return
	}
	if err := w.svc.storage.Delete(ctx, o.StorageKey); err != nil {
		slog.Warn("delete object blob", "object", o.ID, "error", err)
		return
	}
	if err := w.svc.store.DeleteObjectRow(ctx, o.ID); err != nil && !errors.Is(err, ErrObjectNotFound) {
		slog.Warn("delete object row", "object", o.ID, "error", err)
	}
}