Thirdshop/Kazier
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ocr is working

Browse Source
This commit is contained in:
m
2026-09-16 12:44:48 +02:00
parent 050434325a
commit 906d48702c
21 changed files with 595 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/cmd/server/main.go
+32 -4
View File
@@ -1,9 +1,15 @@
package main
import (
"context"
"errors"
"fmt"
"log"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"github.com/gin-gonic/gin"
"github.com/vaultdrop/backend/config"
@@ -64,8 +70,30 @@ func main() {
)
handlers.Ocr = service.NewOcr(repo, cfg.UploadDir, cfg.OcrLang, ocr.NewTesseract())
if err := newRouter().Run(fmt.Sprintf(":%d", cfg.Port)); err != nil {
log.Fatalln(err)
}
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
}
// Worker OCR borné (OCR_WORKERS) : file `ocr_jobs`, hors request handlers.
go handlers.Ocr.Run(ctx, cfg.OcrWorkers)
log.Printf("ocr worker démarré (%d workers)", cfg.OcrWorkers)
srv := &http.Server{
Addr: fmt.Sprintf(":%d", cfg.Port),
Handler: newRouter(),
}
go func() {
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("server: %v", err)
}
}()
<-ctx.Done()
stop()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := srv.Shutdown(shutdownCtx); err != nil {
log.Printf("shutdown: %v", err)
}
log.Println("arrêt propre")
}
backend/config/config.go
+7
View File
@@ -13,6 +13,7 @@ type ApplicationConfig struct {
UploadDir string
MaxFileSizeMB int64
OcrLang string
OcrWorkers int
AuthSecret string
AdminUsername string
AdminPassword string
@@ -33,12 +34,18 @@ func LoadApplicationConfig() (*ApplicationConfig, error) {
return nil, err
}
ocrWorkers, err := getInt("OCR_WORKERS", 2)
if err != nil {
return nil, err
}
return &ApplicationConfig{
Port: port,
DatabaseURL: get("DATABASE_URL", "postgres://vaultdrop:vaultdrop@localhost:5432/vaultdrop_dev?sslmode=disable"),
UploadDir: get("UPLOAD_DIR", "./uploads"),
MaxFileSizeMB: int64(maxSize),
OcrLang: get("OCR_LANG", "fra+eng"),
OcrWorkers: ocrWorkers,
AuthSecret: get("AUTH_SECRET", "dev-secret-change-me"),
AdminUsername: get("ADMIN_USERNAME", ""),
AdminPassword: get("ADMIN_PASSWORD", ""),
backend/config/config_test.go
+6 -2
View File
@@ -9,7 +9,7 @@ func clearEnv(t *testing.T) {
t.Helper()
for _, name := range []string{
"PORT", "DATABASE_URL", "UPLOAD_DIR", "MAX_FILE_SIZE_MB",
"OCR_LANG", "AUTH_SECRET", "ADMIN_USERNAME", "ADMIN_PASSWORD",
"OCR_LANG", "OCR_WORKERS", "AUTH_SECRET", "ADMIN_USERNAME", "ADMIN_PASSWORD",
} {
t.Setenv(name, "")
}
@@ -37,6 +37,9 @@ func TestLoadApplicationDefaults(t *testing.T) {
if cfg.OcrLang != "fra+eng" {
t.Errorf("OcrLang défaut inattendu: %q", cfg.OcrLang)
}
if cfg.OcrWorkers != 2 {
t.Errorf("OcrWorkers = %d, attendu 2", cfg.OcrWorkers)
}
if cfg.AuthSecret != "dev-secret-change-me" {
t.Errorf("AuthSecret défaut inattendu: %q", cfg.AuthSecret)
}
@@ -52,6 +55,7 @@ func TestLoadApplicationEnvOverrides(t *testing.T) {
t.Setenv("DATABASE_URL", "postgres://u:p@host:5433/db?sslmode=disable")
t.Setenv("UPLOAD_DIR", "/tmp/up")
t.Setenv("OCR_LANG", "eng")
t.Setenv("OCR_WORKERS", "4")
t.Setenv("AUTH_SECRET", "super-secret")
t.Setenv("ADMIN_USERNAME", "root")
t.Setenv("ADMIN_PASSWORD", "toor")
@@ -61,7 +65,7 @@ func TestLoadApplicationEnvOverrides(t *testing.T) {
t.Fatalf("LoadApplicationConfig: %v", err)
}
if cfg.Port != 9090 || cfg.MaxFileSizeMB != 120 || cfg.DatabaseURL != "postgres://u:p@host:5433/db?sslmode=disable" ||
cfg.UploadDir != "/tmp/up" || cfg.OcrLang != "eng" || cfg.AuthSecret != "super-secret" ||
cfg.UploadDir != "/tmp/up" || cfg.OcrLang != "eng" || cfg.OcrWorkers != 4 || cfg.AuthSecret != "super-secret" ||
cfg.AdminUsername != "root" || cfg.AdminPassword != "toor" {
t.Errorf("overrides non appliqués: %+v", cfg)
}
backend/handlers/ocr_test.go
+6
View File
@@ -42,6 +42,12 @@ func setupOcr(t *testing.T) (*gin.Engine, string, *repository.Repository) {
handlers.Store = store
handlers.Ocr = service.NewOcr(repo, uploadDir, "fra+eng", stubEngine{text: "HELLO OCR"})
// Worker OCR borné — le handler n'enfile plus que le job, le traitement
// tourne sur un worker de contexte.
ctx, cancel := context.WithCancel(context.Background())
go handlers.Ocr.Run(ctx, 1)
t.Cleanup(cancel)
gin.SetMode(gin.TestMode)
r := gin.New()
handlers.RegisterRoutes(r)
backend/repository/ocr_jobs.go
+47
View File
@@ -3,6 +3,7 @@ package repository
import (
"database/sql"
"errors"
"time"
)
// OcrJobRow maps ocr_jobs.
@@ -18,6 +19,9 @@ type OcrJobRow struct {
// ErrJobNotFound marks an OCR job absent or owned by another device.
var ErrJobNotFound = errors.New("ocr job not found")
// ErrNoQueuedJobs marks an OCR queue drained (no job to claim).
var ErrNoQueuedJobs = errors.New("no queued ocr jobs")
type OcrJobs struct{ DB *sql.DB }
func (o *OcrJobs) Create(jobID, deviceID, fileID string) error {
@@ -62,6 +66,49 @@ func (o *OcrJobs) TouchProcessing(deviceID, jobID string) error {
return err
}
// ClaimNext atomically picks the oldest queued job (FIFO) and moves it to
// `processing`. Safe for concurrent workers: `FOR UPDATE SKIP LOCKED` blocks
// the row as part of the same statement. Empty queue → ErrNoQueuedJobs.
func (o *OcrJobs) ClaimNext() (OcrJobRow, error) {
var row OcrJobRow
err := o.DB.QueryRow(
`WITH next AS (
SELECT job_id FROM ocr_jobs
WHERE status = 'queued'
ORDER BY created_at, job_id
LIMIT 1
FOR UPDATE SKIP LOCKED
)
UPDATE ocr_jobs
SET status = 'processing', started_at = NOW()
FROM next
WHERE ocr_jobs.job_id = next.job_id
RETURNING ocr_jobs.job_id, ocr_jobs.file_id, ocr_jobs.device_id, ocr_jobs.status`,
).Scan(&row.ID, &row.FileID, &row.DeviceID, &row.Status)
if err == sql.ErrNoRows {
return OcrJobRow{}, ErrNoQueuedJobs
}
if err != nil {
return OcrJobRow{}, err
}
return row, nil
}
// ResetStaleProcessing requeues jobs stuck in `processing` (worker crash /
// processus redémarré) et plus vieux que `olderThan`. Retourne le nb de jobs
// requeued.
func (o *OcrJobs) ResetStaleProcessing(olderThan time.Duration) (int64, error) {
res, err := o.DB.Exec(
`UPDATE ocr_jobs SET status = 'queued', started_at = NULL
WHERE status = 'processing' AND started_at < $1`,
time.Now().Add(-olderThan),
)
if err != nil {
return 0, err
}
return res.RowsAffected()
}
func (o *OcrJobs) Complete(deviceID, jobID, text string) error {
_, err := o.DB.Exec(
`UPDATE ocr_jobs SET status = 'done', text = NULLIF($3, ''), started_at = COALESCE(started_at, NOW()), completed_at = NOW()
backend/repository/ocr_jobs_test.go
+89
View File
@@ -3,6 +3,7 @@ package repository
import (
"errors"
"testing"
"time"
"github.com/vaultdrop/backend/dbtest"
)
@@ -90,3 +91,91 @@ func TestOcrJobsGetScopedByDevice(t *testing.T) {
t.Errorf("get par un autre device : attendu ErrJobNotFound, got %v", err)
}
}
func TestOcrJobsClaimNextFifo(t *testing.T) {
repo, _, deviceID, fileID := seedOcrJobFixture(t)
var jobIDs []string
for i := 0; i < 3; i++ {
jobID := NewID()
if err := repo.OcrJobs.Create(jobID, deviceID, fileID); err != nil {
t.Fatalf("create %d: %v", i, err)
}
jobIDs = append(jobIDs, jobID)
}
for i, want := range jobIDs {
row, err := repo.OcrJobs.ClaimNext()
if err != nil {
t.Fatalf("claim %d: %v", i, err)
}
if row.ID != want {
t.Errorf("claim %d = %q, attendu %q (FIFO)", i, row.ID, want)
}
if row.Status != "processing" {
t.Errorf("claim %d status = %q, attendu processing", i, row.Status)
}
}
if _, err := repo.OcrJobs.ClaimNext(); !errors.Is(err, ErrNoQueuedJobs) {
t.Errorf("queue vide : attendu ErrNoQueuedJobs, got %v", err)
}
}
func TestOcrJobsClaimNextSkipsProcessingAndDone(t *testing.T) {
repo, _, deviceID, fileID := seedOcrJobFixture(t)
job1 := NewID()
job2 := NewID()
if err := repo.OcrJobs.Create(job1, deviceID, fileID); err != nil {
t.Fatalf("create job1: %v", err)
}
if err := repo.OcrJobs.Create(job2, deviceID, fileID); err != nil {
t.Fatalf("create job2: %v", err)
}
if err := repo.OcrJobs.TouchProcessing(deviceID, job2); err != nil {
t.Fatalf("touch job2 processing: %v", err)
}
row, err := repo.OcrJobs.ClaimNext()
if err != nil {
t.Fatalf("claim: %v", err)
}
if row.ID != job1 {
t.Errorf("claim = %q, attendu job1 %q", row.ID, job1)
}
}
func TestOcrJobsResetStaleProcessing(t *testing.T) {
repo, _, deviceID, fileID := seedOcrJobFixture(t)
jobID := NewID()
if err := repo.OcrJobs.Create(jobID, deviceID, fileID); err != nil {
t.Fatalf("create: %v", err)
}
if err := repo.OcrJobs.TouchProcessing(deviceID, jobID); err != nil {
t.Fatalf("touch processing: %v", err)
}
// Started_at = NOW : pas encore stale → aucun requeue.
if n, err := repo.OcrJobs.ResetStaleProcessing(time.Minute); err != nil {
t.Fatalf("reset: %v", err)
} else if n != 0 {
t.Errorf("reset récent : attendu 0, got %d", n)
}
// Vieillit le started_at puis re-reset → 1 requeue.
if _, err := repo.OcrJobs.DB.Exec(`UPDATE ocr_jobs SET started_at = NOW() - interval '2 minutes' WHERE job_id = $1`, jobID); err != nil {
t.Fatalf("vieillir started_at: %v", err)
}
if n, err := repo.OcrJobs.ResetStaleProcessing(time.Minute); err != nil {
t.Fatalf("reset stale: %v", err)
} else if n != 1 {
t.Errorf("reset stale : attendu 1, got %d", n)
}
row, err := repo.OcrJobs.Get(deviceID, jobID)
if err != nil {
t.Fatalf("get: %v", err)
}
if row.Status != "queued" {
t.Errorf("status après reset = %q, attendu queued", row.Status)
}
}
backend/repository/resources.go
+15
View File
@@ -254,6 +254,21 @@ func (r *Resources) GetFileVisible(userID, resourceID string) (FileRow, error) {
return file, err
}
// GetResourceOwner returns the owning user_id of any resource (file or folder).
// No-rows → ErrNotFound.
func (r *Resources) GetResourceOwner(resourceID string) (string, error) {
var ownerID string
err := r.DB.QueryRow(
`SELECT user_id FROM resources
WHERE resource_id = $1 AND deleted_at IS NULL`,
resourceID,
).Scan(&ownerID)
if errors.Is(err, sql.ErrNoRows) {
return "", ErrNotFound
}
return ownerID, err
}
// DeleteFile soft-deletes the file (deleted_at), returning its id.
func (r *Resources) DeleteFile(ownerID, resourceID string) (string, error) {
result, err := r.DB.Exec(
backend/service/ocr.go
+73 -15
View File
@@ -2,13 +2,25 @@ 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"`
@@ -17,22 +29,24 @@ type OcrJobDTO struct {
Error string `json:"error,omitempty"`
}
// Ocr queues OCR jobs and processes them asynchronously (V1 : goroutine par
// job ; le client poll GET /ocr/jobs/:id toutes les 3s).
// 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}
return &Ocr{Repository: repo, UploadDir: uploadDir, Lang: lang, Engine: engine, StaleAfter: StaleClaimTimeout}
}
// Create valide la ressource (ownership par user), met le job en file et
// lance le traitement. Le job reste scopé par device (le service génère les
// jobs du device courant ; l'outbox OCR n'est pas synchronisée entre devices).
// 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
@@ -41,7 +55,6 @@ func (o *Ocr) Create(userID, deviceID, fileID string) (OcrJobDTO, error) {
if err := o.Repository.OcrJobs.Create(jobID, deviceID, fileID); err != nil {
return OcrJobDTO{}, err
}
go o.process(userID, deviceID, jobID, fileID)
return OcrJobDTO{ID: jobID, Status: "queued"}, nil
}
@@ -53,22 +66,67 @@ func (o *Ocr) Get(deviceID, jobID string) (OcrJobDTO, error) {
return toOcrJobDTO(row), nil
}
func (o *Ocr) process(userID, deviceID, jobID, fileID string) {
ctx := context.Background()
if err := o.Repository.OcrJobs.TouchProcessing(deviceID, jobID); err != 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, fileID)
path, err := o.physicalPath(userID, row.FileID)
if err != nil {
_ = o.Repository.OcrJobs.Fail(deviceID, jobID, "file not readable")
_ = 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(deviceID, jobID, err.Error())
_ = o.Repository.OcrJobs.Fail(row.DeviceID, row.ID, err.Error())
return
}
_ = o.Repository.OcrJobs.Complete(deviceID, jobID, text)
_ = o.Repository.OcrJobs.Complete(row.DeviceID, row.ID, text)
}
// physicalPath résout UPLOAD_DIR/<user_id>/<resource_id>.<ext> — l'ext est
@@ -95,4 +153,4 @@ func toOcrJobDTO(row repository.OcrJobRow) OcrJobDTO {
dto.Error = *row.Error
}
return dto
}
}
backend/service/ocr_test.go
+7 -2
View File
@@ -26,13 +26,18 @@ func (s stubEngine) ExtractText(_ context.Context, _ string, _ string) (string,
return s.text, nil
}
// newTestOcr builds an Ocr over a fresh DB and returns it with the ids.
// newTestOcr builds an Ocr over a fresh DB, starts the bounded worker (1
// worker) and returns it with the ids.
func newTestOcr(t *testing.T, uploadDir string, engine ocr.Engine) (*Ocr, string, string, string) {
t.Helper()
s := newServiceStore(t)
userID := mustCreateUser(t, s.Repository, "ocr-failed")
deviceID := mustRegisterDevice(t, s.Repository, repository.NewID())
return NewOcr(s.Repository, uploadDir, "fra+eng", engine), userID, deviceID, uploadDir
o := NewOcr(s.Repository, uploadDir, "fra+eng", engine)
ctx, cancel := context.WithCancel(context.Background())
go o.Run(ctx, 1)
t.Cleanup(cancel)
return o, userID, deviceID, uploadDir
}
// waitTillTerminal poll jusqu'à un statut terminal (done/failed).