Kazier/backend/service/ocr.go

package service

import (
	"context"
	"errors"
	"log"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/vaultdrop/backend/ocr"
	"github.com/vaultdrop/backend/repository"
)

const (
	// StaleClaimTimeout : un job `processing` plus vieux que ça est ré-enfilé
	// au boot (worker mort / process redémarré).
	StaleClaimTimeout = 15 * time.Minute
	// QueuePollInterval : cadence de scrutation quand la file est vide.
	QueuePollInterval = time.Second
)

// OcrJobDTO serializes exactly as mobile/api/types.ts OcrJob.
type OcrJobDTO struct {
	ID     string `json:"id"`
	Status string `json:"status"`
	Text   string `json:"text,omitempty"`
	Error  string `json:"error,omitempty"`
}

// Ocr queues OCR jobs in `ocr_jobs` (status `queued`) and processes them
// asynchronously through a bounded worker ([Run]) — Tesseract ne tourne jamais
// dans un request handler. Le client poll GET /ocr/jobs/:id toutes les 3s.
type Ocr struct {
	Repository *repository.Repository
	UploadDir  string
	Lang       string
	Engine     ocr.Engine
	StaleAfter time.Duration
}

func NewOcr(repo *repository.Repository, uploadDir, lang string, engine ocr.Engine) *Ocr {
	return &Ocr{Repository: repo, UploadDir: uploadDir, Lang: lang, Engine: engine, StaleAfter: StaleClaimTimeout}
}

// Create valide la ressource (ownership par user) et l'enfile uniquement
// (status `queued`) — le traitement est délégué au worker [Run] pour ne pas
// stresser le serveur. Le job reste scopé par device.
func (o *Ocr) Create(userID, deviceID, fileID string) (OcrJobDTO, error) {
	if _, err := o.Repository.Resources.GetFile(userID, fileID); err != nil {
		return OcrJobDTO{}, err
	}
	jobID := repository.NewID()
	if err := o.Repository.OcrJobs.Create(jobID, deviceID, fileID); err != nil {
		return OcrJobDTO{}, err
	}
	return OcrJobDTO{ID: jobID, Status: "queued"}, nil
}

func (o *Ocr) Get(deviceID, jobID string) (OcrJobDTO, error) {
	row, err := o.Repository.OcrJobs.Get(deviceID, jobID)
	if err != nil {
		return OcrJobDTO{}, err
	}
	return toOcrJobDTO(row), nil
}

// Run fait tourner le worker OCR : au démarrage il requeue les jobs laissés
// en `processing` (crash), puis `workers` boucles consomment la file FIFO via
// ClaimNext (atomique, `FOR UPDATE SKIP LOCKED`). `workers` borne la
// concurrence Tesseract. Retourne quand ctx est annulé.
func (o *Ocr) Run(ctx context.Context, workers int) {
	if workers < 1 {
		workers = 1
	}
	if _, err := o.Repository.OcrJobs.ResetStaleProcessing(o.StaleAfter); err != nil {
		log.Printf("ocr: requeue jobs stale: %v", err)
	}

	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			o.workerLoop(ctx)
		}()
	}
	wg.Wait()
}

// workerLoop consomme la file : un job `queued` → processing → done/failed.
// File vide → scrutation à QueuePollInterval jusqu'à annulation du ctx.
func (o *Ocr) workerLoop(ctx context.Context) {
	for {
		row, err := o.Repository.OcrJobs.ClaimNext()
		if err != nil {
			if !errors.Is(err, repository.ErrNoQueuedJobs) {
				log.Printf("ocr: claim: %v", err)
			}
			select {
			case <-ctx.Done():
				return
			case <-time.After(QueuePollInterval):
			}
			continue
		}
		o.process(ctx, row)
	}
}

// process exécute l'extraction pour un job déjà réclamé (`processing`).
func (o *Ocr) process(ctx context.Context, row repository.OcrJobRow) {
	userID, err := o.Repository.Resources.GetResourceOwner(row.FileID)
	if err != nil {
		_ = o.Repository.OcrJobs.Fail(row.DeviceID, row.ID, "file not readable")
		return
	}
	path, err := o.physicalPath(userID, row.FileID)
	if err != nil {
		_ = o.Repository.OcrJobs.Fail(row.DeviceID, row.ID, "file not readable")
		return
	}
	text, err := o.Engine.ExtractText(ctx, path, o.Lang)
	if err != nil {
		_ = o.Repository.OcrJobs.Fail(row.DeviceID, row.ID, err.Error())
		return
	}
	_ = o.Repository.OcrJobs.Complete(row.DeviceID, row.ID, text)
	// Le texte atterrit sur la ressource (updated_at bumpé) : consultable par
	// tous les devices du user et les grantees via GET /files/:id/ocr. SILENCIEUX
	// (hôte déjà supprimé ou lock d'écriture) : le job reste source d'historique.
	if trimmed := strings.TrimSpace(text); trimmed != "" {
		if err := o.Repository.Resources.UpdateOcrText(userID, row.FileID, trimmed); err != nil {
			log.Printf("ocr: persiste ocr_text de %s: %v", row.FileID, err)
		}
	}
}

// physicalPath résout UPLOAD_DIR/<user_id>/<resource_id>.<ext> — l'ext est
// choisi à l'upload, le fichier réel est retrouvé par préfixe.
func (o *Ocr) physicalPath(userID, fileID string) (string, error) {
	matches, err := filepath.Glob(filepath.Join(o.UploadDir, userID, fileID+".*"))
	if err != nil {
		return "", err
	}
	for _, m := range matches {
		if strings.HasPrefix(filepath.Base(m), fileID+".") {
			return m, nil
		}
	}
	return "", repository.ErrNotFound
}

func toOcrJobDTO(row repository.OcrJobRow) OcrJobDTO {
	dto := OcrJobDTO{ID: row.ID, Status: row.Status}
	if row.Text != nil {
		dto.Text = *row.Text
	}
	if row.Error != nil {
		dto.Error = *row.Error
	}
	return dto
}