# seoinsight v2 — Socle jobs (T0) : table, worker, GET /jobs/{id}

Date : 2026-08-31
Statut : **validée par l'utilisateur le 2026-08-31** (« Go valide les deux specs ») — T0 peut commencer.
Amendé le 2026-08-31 suite à review (voir `reviews/2026-08-31-imports-specs-review.md`).
S'appuie sur : architecture (2026-08-26 §7), data (2026-08-28 §5), module Imports (2026-08-31 §9 T0).

Socle **neutre** dans `api/app/jobs/` : il n'appartient ni à Crawl ni à Imports, il débloque les deux. Techno actée (décision utilisateur 2026-08-31) : **polling Postgres `FOR UPDATE SKIP LOCKED`, sans broker** — zéro infra en plus sur la Dedibox. Une feature, une migration (`jobs`), enregistrée dans `api/app/db/models.py`.

## 1. Table `jobs` (DATA)

| Colonne | Type | Contraintes |
|---|---|---|
| `id` | `UUID` | PK, `server_default=uuidv7()` |
| `org_id` | `TEXT` | NOT NULL, indexé |
| `type` | `TEXT` | NOT NULL — `import_csv`, `import_api`, `map_reconcile`, plus tard `crawl`, … |
| `status` | `TEXT` | NOT NULL, `CHECK (status IN ('queued','running','done','failed'))`, défaut `queued` |
| `attempts` | `INTEGER` | NOT NULL, défaut 0 — incrémenté à chaque claim (`queued`→`running`) ; plafond `max_attempts` déclaré **par type** dans le registre (défaut 3, architecture §9) |
| `dedup_key` | `TEXT` | **UNIQUE, nullable** — colonne dédiée, jamais d'index UNIQUE sur jsonb ; moment du relâchement déclaré **par type** (`dedup_scope`, voir ci-dessous) |
| `payload` | `JSONB` | NOT NULL, défaut `{}` — arguments du job ; convention : `site_id`/`site_target_id` au premier niveau quand le job est scopé |
| `progress` | `JSONB` | nullable — écrit par le job en cours d'exécution |
| `report` | `JSONB` | nullable — écrit une fois, au passage à `done`/`failed` |
| `error` | `TEXT` | nullable — message court du `failed` (le détail va dans `report`) |
| `created_at` / `started_at` / `finished_at` | `TIMESTAMPTZ` | `created_at` NOT NULL défaut `now()` ; les deux autres nullables |

Index : `(org_id, type, created_at DESC)` ; index partiel `(status, created_at)` `WHERE status IN ('queued','running')` (file d'attente et surveillance — la table grossit, la file reste petite).

Rétention : `jobs` n'est jamais purgée en v1. Les événements de l'onglet Imports (spec Imports, écart n°3) sont des lignes `map_reconcile` de cette table : toute purge future devra les épargner ou les archiver — dépendance actée.

**`dedup_key`** : anti-empilement, pas anti-rejeu. À l'enfilage : `INSERT … ON CONFLICT (dedup_key) DO NOTHING` + relecture par `dedup_key` si conflit (convention `"<type>:<scope>"`, ex. `map_reconcile:<target_id>`). **La portée de la dédup est tranchée par type**, déclarée avec le handler dans le registre (`dedup_scope`) :

- **`'queue'`** (jobs idempotents et bon marché — `map_reconcile`) : le claim (`queued`→`running`) **remet `dedup_key` à NULL au démarrage du run**. Un même travail ne s'empile jamais *en file*, mais redevient enfilable dès que le run démarre : un crawl B qui finit pendant un run de reconcile enfile un nouveau job, exécuté ensuite — **zéro lost-wakeup**, l'invariant de convergence (Imports §3.9) tient quel que soit l'ordre d'arrivée. Deux runs concurrents (à N workers) restent sûrs : le handler est un UPDATE ensembliste idempotent.
- **`'active'`** (jobs longs et exclusifs — `crawl`) : la clé n'est relâchée qu'au passage `done`/`failed` : jamais deux instances actives du même travail — la garde « un seul crawl actif par site » reste vraie même à N workers.

Course résiduelle à l'enfilage : le gagnant peut finir (et NULLer sa clé) entre le conflit et la relecture → relecture vide → **ré-insérer** (boucle bornée, 3 essais) au lieu de renvoyer une erreur.

**`progress`** : forme libre par type, avec deux clés conventionnelles : `updated_at` (heartbeat) et, si dénombrable, `done`/`total`. Écrit par lot/étape, jamais à chaque ligne. **`report`** : résultat final structuré (pour `import_csv` : lues/insérées/ignorées/erreurs — spec Imports §4).

## 2. Worker (`uv run python -m app.jobs.worker`)

Un process, un job à la fois (suffisant pour les imports ; voir §4 pour la montée).

1. Boucle : `SELECT … FROM jobs WHERE status='queued' ORDER BY created_at LIMIT 1 FOR UPDATE SKIP LOCKED` ; rien → sleep court (1–2 s) ; sinon `status='running'`, `started_at=now()`, `attempts=attempts+1`, `dedup_key=NULL` si le type est `dedup_scope='queue'`, commit, exécution.
2. Dispatch par `type` vers un registre de handlers (`app/jobs/registry.py` ; chaque module enregistre les siens — Imports enregistrera `import_csv` et `map_reconcile`). Une entrée du registre déclare : le handler, `max_attempts` (défaut 3), `dedup_scope` (défaut `'queue'`), et un **callback optionnel `on_fail(job)`** possédé par le module — invoqué (best-effort, erreurs loguées) après tout passage à `failed`, par le worker comme par le sweep, pour que le module mette son état de domaine en cohérence (Imports §3.11) : le worker reste neutre, il ne connaît que le registre.
3. Fin : `done` + `report`, ou `failed` + `error`/`report` ; `finished_at=now()`, `dedup_key=NULL` (déjà NULL pour `dedup_scope='queue'`). Une exception non attrapée = `failed` immédiat et définitif (échec déterministe, le rejeu automatique n'aiderait pas), jamais un worker mort ; `on_fail` invoqué.
4. **Arrêt propre** : SIGTERM/SIGINT → cesse de prendre des jobs, termine le job courant, sort. Les jobs sont idempotents par contrat (Imports §4) : un kill -9 est rattrapé par le point 5.
5. **Jobs zombies** (worker mort à mi-course : kill -9, crash natif — un CSV qui fait segfaulter DuckDB tue le process sans exception Python —, machine redémarrée) : au démarrage puis périodiquement, le sweep prend tout job `running` dont `progress->>'updated_at'` (à défaut `started_at`) est plus vieux qu'un seuil (défaut 10 min) : si `attempts < max_attempts` → repasse `queued` (le prochain claim ré-incrémente `attempts`), le rejeu est sûr par idempotence ; **sinon → `failed` définitif** (`error='max attempts exceeded'`, `finished_at=now()`, `dedup_key=NULL`, `on_fail` invoqué). Un job empoisonné ne monopolise jamais le worker : la file continue (architecture §9).

## 3. API

`GET /jobs/{id}` — `CurrentOrgDep`, filtre `org_id`, 404 si étranger à l'org. Réponse : `{id, type, status, progress, report, error, created_at, started_at, finished_at}`. C'est l'unique endpoint du socle ; le polling web (composant partagé, spec Imports §7) ne consomme que lui.

## 4. Besoins futurs vérifiés (ce que Crawl exigera — porté sans migration)

| Besoin | Comment la table le porte, sans changer de schéma |
|---|---|
| **Concurrence par site** | Montée = N process worker identiques : `SKIP LOCKED` répartit nativement. « Un seul crawl actif par site » = `dedup_key = "crawl:<site_id>"` en **`dedup_scope='active'`** (clé tenue jusqu'à `done`/`failed` : bloque l'empilement en file *et* la double exécution) + garde dans la requête de claim : exclure les jobs dont `payload->>'site_id'` a déjà un job `running` de même type. `payload` est requêtable ; un index d'expression sur `(payload->>'site_id')` reste possible sans changement de colonnes. |
| **Rate limiting** | Intra-job : le crawl d'un site est UN job long qui cadence lui-même ses requêtes httpx (jamais un job par URL) — la table n'a rien à porter. Une cadence inter-jobs se ferait par la même garde de claim sur `payload`. |
| **Reprise longue** | `progress` = checkpoint (curseur de frontière, run_id, compteurs) relu par le handler au rejeu ; heartbeat `updated_at` + balayage des zombies (§2.5) ; le contrat d'idempotence rend le rejeu partiel sûr. Testé dès T2 sur `import_csv` (reprise à mi-course sans doublon). |
| **Priorités** (non exigé) | `ORDER BY created_at` suffit ; au besoin, `payload->>'priority'` dans l'ORDER BY du claim. Si un vrai besoin apparaît, une colonne `priority` est une migration additive triviale — assumé. |

Limite assumée (cas résiduel 10 de la synthèse) : la preuve définitive n'arrivera qu'au design du module Crawl ; si le polling Postgres ne suffisait pas (improbable aux volumes visés), la migration serait d'infra (broker), pas de schéma.

## 5. Tests

`test_jobs_worker.py` : claim exclusif à deux workers concurrents (SKIP LOCKED, aucun job pris deux fois) ; `dedup_key` — deuxième enfilage rend le job existant ; **scénario A/B anti-lost-wakeup** (`dedup_scope='queue'`) : job `running`, ré-enfilage accepté, exécuté ensuite ; `dedup_scope='active'` : ré-enfilage refusé tant que `running`, possible après `done` ; **fallback de course** : gagnant terminé entre conflit et relecture → ré-insertion réussie ; zombie requalifié `queued` avec `attempts` incrémenté ; **zombie au plafond (3) → `failed` définitif, la file continue** (le job suivant est pris) ; échec handler → `failed` + `error` + `on_fail` invoqué, worker vivant ; `GET /jobs/{id}` filtré par org (404 étranger).

## Écarts

Aucun écart restant à l'architecture après amendement du 2026-08-31 : `attempts`/`max_attempts` et le `failed` définitif (§1, §2.5) alignent le socle sur architecture §9 (manque relevé en review, C2) ; le nom du process est aligné sur architecture §7 (**`python -m app.jobs.worker`**, préférence du reviewer, tranché au plus simple — pas de module à la racine de `app/`).