Two-stage processing

<?php

declare(strict_types=1);

namespace App\Upload;

use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\HttpFoundation\{Request, Response, JsonResponse};
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Routing\Annotation\Route;

final class UploadController extends AbstractController
{
    public function __construct(
        private readonly MessageBusInterface $bus,
        private readonly JobRepository $jobs,
        private readonly BlobStorage $blob,
    ) {}

    #[Route('/uploads', methods: ['POST'])]
    public function upload(Request $req): Response
    {
        // Stage 1: fast ack
        $file = $req->files->get('file');
        if (!$file) {
            return new JsonResponse(['error' => 'file missing'], 400);
        }

        // Persist raw bytes to object storage (S3/minio)
        $blobKey = $this->blob->put($file->getPathname(), $file->getMimeType());

        // Persist job in DB in queued state
        $jobId = bin2hex(random_bytes(16));
        $this->jobs->create([
            'id'     => $jobId,
            'kind'   => 'video.transcode',
            'status' => 'queued',
            'input'  => [
                'blobKey' => $blobKey,
                'mime'    => $file->getMimeType(),
                'size'    => $file->getSize(),
            ],
        ]);

        // Dispatch async message (ideally inside outbox-style TX with job insert)
        $this->bus->dispatch(new TranscodeVideo($jobId));

        return new JsonResponse(
            ['jobId' => $jobId, 'status' => 'queued'],
            Response::HTTP_ACCEPTED,
            ['Location' => "/jobs/$jobId", 'Retry-After' => '5'],
        );
    }

    #[Route('/jobs/{id}', methods: ['GET'])]
    public function status(string $id): JsonResponse
    {
        $job = $this->jobs->find($id);
        if (!$job) {
            return new JsonResponse(['error' => 'not found'], 404);
        }

        return new JsonResponse([
            'id'       => $job['id'],
            'status'   => $job['status'],
            'progress' => $job['progress'],
            'result'   => $job['result'],
            'error'    => $job['error'],
        ]);
    }
}

<?php

declare(strict_types=1);

namespace App\Upload;

use Symfony\Component\Messenger\Attribute\AsMessageHandler;

final readonly class TranscodeVideo
{
    public function __construct(public string $jobId) {}
}

#[AsMessageHandler]
final class TranscodeVideoHandler
{
    public function __construct(
        private JobRepository $jobs,
        private BlobStorage $blob,
        private Transcoder $transcoder,
        private ThumbnailService $thumb,
    ) {}

    public function __invoke(TranscodeVideo $msg): void
    {
        $job = $this->jobs->find($msg->jobId);
        if (!$job || $job['status'] === 'done') {
            return; // idempotent
        }

        $this->jobs->markStarted($msg->jobId);

        try {
            $input = $job['input'];

            // Heavy: stream blob, run ffmpeg, upload outputs
            $outputs = $this->transcoder->transcode(
                sourceKey: $input['blobKey'],
                resolutions: ['144p', '360p', '720p', '1080p'],
                onProgress: fn(int $pct) => $this->jobs->updateProgress($msg->jobId, $pct),
            );

            $thumbKey = $this->thumb->generate($input['blobKey']);

            $this->jobs->markDone($msg->jobId, [
                'outputs' => $outputs,
                'thumbnail' => $thumbKey,
            ]);
        } catch (\Throwable $e) {
            $this->jobs->markFailed($msg->jobId, $e->getMessage());
            // Re-throw so Messenger retries per configured strategy
            throw $e;
        }
    }
}

package upload

import (
	"context"
	"encoding/json"
	"net/http"

	"github.com/google/uuid"
)

type Server struct {
	jobs   JobRepo
	blob   BlobStore
	queue  chan<- Job // producer side of pipeline
}

// POST /uploads -- stage 1: accept file, persist blob + job, enqueue.
func (s *Server) Upload(w http.ResponseWriter, r *http.Request) {
	r.Body = http.MaxBytesReader(w, r.Body, 500<<20) // 500 MB limit

	file, header, err := r.FormFile("file")
	if err != nil {
		http.Error(w, "file missing", http.StatusBadRequest)
		return
	}
	defer file.Close()

	blobKey, err := s.blob.Put(r.Context(), file, header.Header.Get("Content-Type"))
	if err != nil {
		http.Error(w, "storage error", http.StatusInternalServerError)
		return
	}

	jobID := uuid.NewString()
	job := Job{
		ID:     jobID,
		Kind:   "video.transcode",
		Status: "queued",
		Input:  Input{BlobKey: blobKey, Size: header.Size},
	}
	if err := s.jobs.Create(r.Context(), job); err != nil {
		http.Error(w, "db error", http.StatusInternalServerError)
		return
	}

	// Enqueue for stage 2
	select {
	case s.queue <- job:
	case <-r.Context().Done():
		return
	}

	w.Header().Set("Location", "/jobs/"+jobID)
	w.Header().Set("Retry-After", "5")
	w.WriteHeader(http.StatusAccepted)
	_ = json.NewEncoder(w).Encode(map[string]string{"jobId": jobID, "status": "queued"})
}

package upload

import (
	"context"
	"log/slog"
	"sync"
)

// Pipeline processes jobs in stages via channels.
// Each stage is a pool of goroutines fed from a channel.
type Pipeline struct {
	jobs       JobRepo
	transcoder Transcoder
	thumbnailer ThumbnailService
	log        *slog.Logger
}

// Run starts the pipeline. It returns the producer channel (close to stop).
func (p *Pipeline) Run(ctx context.Context, workers int) chan<- Job {
	jobs := make(chan Job, 100)
	transcoded := make(chan transcodeResult, 100)

	// Stage A: transcoding
	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for job := range jobs {
				p.runTranscode(ctx, job, transcoded)
			}
		}(i)
	}

	// Stage B: thumbnailing (fewer workers, cpu-bound but fast)
	var wg2 sync.WaitGroup
	for i := 0; i < workers/2+1; i++ {
		wg2.Add(1)
		go func() {
			defer wg2.Done()
			for res := range transcoded {
				p.runThumbnail(ctx, res)
			}
		}()
	}

	// Cleanup: when producer closes jobs, close transcoded after stage A drains.
	go func() {
		wg.Wait()
		close(transcoded)
		wg2.Wait()
	}()

	return jobs
}

type transcodeResult struct {
	Job     Job
	Outputs []string
}

func (p *Pipeline) runTranscode(ctx context.Context, job Job, out chan<- transcodeResult) {
	_ = p.jobs.MarkStarted(ctx, job.ID)
	outputs, err := p.transcoder.Transcode(ctx, job.Input.BlobKey, []string{"360p", "720p", "1080p"})
	if err != nil {
		_ = p.jobs.MarkFailed(ctx, job.ID, err.Error())
		p.log.Error("transcode failed", "jobID", job.ID, "err", err)
		return
	}
	select {
	case out <- transcodeResult{Job: job, Outputs: outputs}:
	case <-ctx.Done():
	}
}

func (p *Pipeline) runThumbnail(ctx context.Context, r transcodeResult) {
	thumb, err := p.thumbnailer.Generate(ctx, r.Job.Input.BlobKey)
	if err != nil {
		_ = p.jobs.MarkFailed(ctx, r.Job.ID, err.Error())
		return
	}
	_ = p.jobs.MarkDone(ctx, r.Job.ID, Result{Outputs: r.Outputs, Thumbnail: thumb})
}