# SERP Semantics Pipeline (S3) Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Le pipeline asynchrone qui exécute une analyse sémantique SERP de bout en bout : action Next → job `serp-semantics` → worker analytics (fetch mobile, staging R2, wasm semantic-core, exports JSON Lines) → profil persisté dans `keywords.semantic`. **Architecture:** Design Phase B D2-D4 (`docs/features/analyse-semantique-serp/design-technique-phase-b.md`). Pattern job existant (`analytics_jobs` + worker Hono/Bun), runner en injection de dépendances (mirror `src/lib/pages/crawl-runner.ts` / `services/analytics/src/site-crawl.ts`), moteur = wrapper `src/lib/semantic/engine.ts` (S1+S2, branche `ytg-like/semantic-core`). Postgres = vérité (profil jsonb par-locale), R2 = staging bruts (delete app-level + lifecycle) et exports full-grain. **Tech Stack:** Drizzle/Postgres, Bun/Hono (service analytics), R2 (S3-compat), wasm `semantic-core`. ## Global Constraints - Branche **`ytg-like/serp-pipeline`** (empilée sur `ytg-like/semantic-core`). AUCUN merge dans `dev` — validation globale utilisateur en fin de feature. - **Symétrie de config wasm OBLIGATOIRE** : la même `Partial` passe à `extractSemanticPage` ET `buildSerpProfile` (JSDoc engine.ts — scores silencieusement faux sinon). - **Positions clampées 1-10** : le runner filtre toute position < 1 ou > 10 AVANT d'appeler le moteur (discontinuité W_P pos > 10 — décision revue finale). - Écriture profil : **merge SQL jsonb par-locale** `semantic = coalesce(semantic,'{}'::jsonb) || jsonb_build_object($country, $profile)` (pattern exact `src/lib/keywords/serp-store.ts`) — jamais de read-modify-write. - Tenancy fail-closed : `assertKeywordInActiveOrg` en tête de toute action (`src/lib/sites/assert-site.ts`). - Exports full-grain : **JSON Lines gzippés** (`pages.jsonl.gz`, `terms.jsonl.gz`) sur R2 sous `serp-semantics-results/` — le CDC §4.4 autorise Parquet OU JSON Lines ; Parquet différé à l'ouverture du read-path DuckDB (décision design D4, documentée ici). - Staging bruts : R2 `serp-semantics-staging/${siteId}/${jobId}/pos-${n}.html.gz`, upload best-effort, **delete app-level en fin de job** (succès OU échec) ; lifecycle R2 en garde-fou (note ops, hors code). - UA mobile par défaut (CDC §3.0) : `Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Mobile Safari/537.36 (compatible; SEOInsightBot/1.0)` — constante `MOBILE_UA`, surchargeable par config. - DB de test : les tests DB utilisent le pattern existant (`assertTestDb`, `DATABASE_URL=...seo_insight_test`, script `test` — jamais `bun test` nu). - Timestamps : le runner reçoit `now: () => string` en dépendance (testabilité) ; `analyzedAt`/`jobId`/`exports` sont ajoutés côté TS au profil (le crate ne les produit pas). - Conventional Commits en français. Gates récurrentes : `bun run test:all` sans régression, `bunx tsc --noEmit` clean. --- ### Task 1: Migration DB — colonne `semantic` + payload de job + type `serp-semantics` **Files:** - Modify: `src/lib/db/domain-schema.ts` (table `keywords` : colonne `semantic` ; table `analyticsJobs` : colonne `payload`, union `type`) - Create: `drizzle/00XX_semantic_profile.sql` (généré par `bun run db:generate`, puis ALTER lz4 appendé) - Create: `src/lib/semantic/profile-types.ts` (type de stockage) **Interfaces:** - Produces: - `keywords.semantic` : `jsonb $type>` (clé = country), nullable ; - `analyticsJobs.payload` : `jsonb $type` ; - `analyticsJobs.type` union étendue : `"csv-import" | "site-crawl" | "serp-semantics"` ; - `StoredSemanticProfile = SemanticProfile & { analyzedAt: string; jobId: string; exports: { pagesKey: string; termsKey: string } }` (dans `profile-types.ts`, importe `SemanticProfile` de `./types`) ; - `SerpSemanticsPayload = { keywordId: string; country: string; urls: { position: number; url: string; included: boolean }[]; clientUrl?: string }`. - [ ] **Step 1: Écrire les types + colonnes** `src/lib/semantic/profile-types.ts` : ```typescript // src/lib/semantic/profile-types.ts — shapes de stockage du profil sémantique // (le profil wasm + les champs ajoutés côté TS : analyzedAt/jobId/exports). import type { SemanticProfile } from "./types"; export type SemanticExports = { pagesKey: string; termsKey: string }; export type StoredSemanticProfile = SemanticProfile & { analyzedAt: string; jobId: string; exports: SemanticExports; }; /** Payload du job serp-semantics (analytics_jobs.payload). */ export type SerpSemanticsPayload = { keywordId: string; country: string; urls: { position: number; url: string; included: boolean }[]; clientUrl?: string; }; ``` `domain-schema.ts` — sur `keywords` (à côté de `serp`) : ```typescript // Profil sémantique SERP par pays (moteur semantic-core). Merge par-locale // comme `serp`/`briefs` ; TOAST lz4 (migration). semantic: jsonb("semantic").$type>(), ``` Sur `analyticsJobs` : étendre le `$type` du champ `type` avec `"serp-semantics"`, et ajouter : ```typescript // Payload structuré des jobs qui ne rentrent pas dans les colonnes plates // (serp-semantics : keywordId, country, urls curées, clientUrl). payload: jsonb("payload").$type(), ``` (Import type only — pas de cycle : `profile-types.ts` n'importe pas le schéma.) - [ ] **Step 2: Générer la migration** Run: `bun run db:generate` Expected: nouvelle migration `drizzle/00XX_*.sql` avec les deux `ALTER TABLE ... ADD COLUMN`. Ouvre-la et appende (mirror `drizzle/0024_keywords_toast_lz4.sql`) : ```sql --> statement-breakpoint ALTER TABLE "keywords" ALTER COLUMN "semantic" SET COMPRESSION lz4; ``` - [ ] **Step 3: Migrer les DB locales** Run: `bun run db:migrate` (dev) puis contre la DB de test (même commande avec `DATABASE_URL` de test, selon le pattern du repo — regarde les scripts `db:*` de package.json). Expected: migrations appliquées sans erreur. - [ ] **Step 4: Vérifier le diff drizzle vide** Run: `bun run db:generate` Expected: « No schema changes » (gate du repo — le partitionnement/compression manuscrits sont invisibles à drizzle-kit, c'est attendu). - [ ] **Step 5: Gates + commit** Run: `bunx tsc --noEmit` → clean. ```bash git add -A && git commit -m "feat(semantic): colonne keywords.semantic (lz4) + payload/type serp-semantics sur analytics_jobs" ``` --- ### Task 2: Fetch mobile-first **Files:** - Create: `src/lib/semantic/fetch.ts` - Test: `src/lib/semantic/fetch.test.ts` **Interfaces:** - Consumes: rien (module feuille ; s'inspire de `src/lib/keywords/onpage-fetch.ts` — lis-le d'abord, mêmes garde-fous). - Produces: `MOBILE_UA` (constante exportée), `fetchSemanticHtml(url: string, fetcher?: typeof fetch, ua?: string): Promise<{ html: string; finalUrl: string } | null>` — timeout 15 s, cap 3 Mo, content-type `text/html`/`xhtml` uniquement, `redirect: "follow"` (le `finalUrl` = `response.url`, consommé par `slug_tokens`), tout échec → `null` (jamais de throw). - [ ] **Step 1: Tests (fetcher injecté, aucun réseau)** ```typescript // src/lib/semantic/fetch.test.ts import { describe, expect, test } from "bun:test"; import { MOBILE_UA, fetchSemanticHtml } from "./fetch"; function fakeFetch(body: string, init: { status?: number; contentType?: string; url?: string } = {}) { return (async (_input: RequestInfo | URL, reqInit?: RequestInit) => { const res = new Response(body, { status: init.status ?? 200, headers: { "content-type": init.contentType ?? "text/html; charset=utf-8" }, }); Object.defineProperty(res, "url", { value: init.url ?? "https://ex.com/final" }); // expose les headers envoyés pour l'assertion UA (globalThis as Record).__lastReqInit = reqInit; return res; }) as typeof fetch; } describe("fetchSemanticHtml", () => { test("renvoie html + finalUrl et envoie l'UA mobile", async () => { const out = await fetchSemanticHtml("https://ex.com/a", fakeFetch("ok")); expect(out?.html).toContain("ok"); expect(out?.finalUrl).toBe("https://ex.com/final"); const init = (globalThis as Record).__lastReqInit as RequestInit; expect(new Headers(init.headers).get("user-agent")).toBe(MOBILE_UA); }); test("null sur non-HTML et sur statut != 2xx", async () => { expect(await fetchSemanticHtml("https://ex.com/a", fakeFetch("{}", { contentType: "application/json" }))).toBeNull(); expect(await fetchSemanticHtml("https://ex.com/a", fakeFetch("nope", { status: 404 }))).toBeNull(); }); test("null quand le fetcher throw", async () => { const throwing = (async () => { throw new Error("boom"); }) as unknown as typeof fetch; expect(await fetchSemanticHtml("https://ex.com/a", throwing)).toBeNull(); }); }); ``` - [ ] **Step 2: Vérifier l'échec** — Run: `bun run test src/lib/semantic/fetch.test.ts` → module absent. - [ ] **Step 3: Implémenter** — mirror `onpage-fetch.ts` (AbortController 15 s, lecture streamée capée 3 Mo, filtre content-type) avec `ua` paramétrable défaut `MOBILE_UA`, et retour `{ html, finalUrl: res.url || url }`. - [ ] **Step 4: Vérifier** — Run: `bun run test src/lib/semantic/fetch.test.ts` → PASS. - [ ] **Step 5: Commit** — `git commit -m "feat(semantic): fetch mobile-first (UA Googlebot-like, 15s, cap 3 Mo)"` --- ### Task 3: Runner en DI — cœur du pipeline **Files:** - Create: `src/lib/semantic/analysis-runner.ts` - Test: `src/lib/semantic/analysis-runner.test.ts` **Interfaces:** - Consumes: `engine.ts` (`extractSemanticPage`, `buildSerpProfile`), `profile-types.ts`, `fetch.ts` (via deps), `mapPool` (cherche l'utilitaire existant utilisé par `src/lib/keywords/serp-actions.ts` — réutilise-le, ne le duplique pas). - Produces (consommé par Task 4) : ```typescript export type RunnerDeps = { db: DrizzleDb; // même type que crawl-runner.ts (regarde son import) fetchHtml: (url: string) => Promise<{ html: string; finalUrl: string } | null>; putObject: (key: string, body: Uint8Array) => Promise; // staging + exports R2 deleteObject: (key: string) => Promise; // best-effort now: () => string; // ISO onProgress?: (p: { phase: "fetch" | "extract" | "score" | "persist"; done: number; total: number }) => Promise; }; export async function runSerpSemantics( jobId: string, siteId: string, payload: SerpSemanticsPayload, deps: RunnerDeps, ): Promise<{ ok: true } | { ok: false; error: string }>; ``` Logique imposée (dans l'ordre) : 1. Charge la ligne `keywords` (`payload.keywordId`) — introuvable → `{ok:false}`. `lang` du mot-clé = langue de référence. 2. **Clamp positions 1-10** + dédup par position. Les URLs `included: false` ne sont PAS fetchées (elles restent dans la liste passée au moteur avec `semantics: null` — le crate les classe `excludedByUser` via `included:false` ; positions non fetchables → `fetchOk:false`). 3. Fetch pool concurrence 4 (`mapPool`) des URLs incluses. Chaque HTML : staging R2 best-effort (`serp-semantics-staging/${siteId}/${jobId}/pos-${n}.html.gz`, gzip via `Bun.gzipSync` ou `node:zlib`), puis `extractSemanticPage(html, finalUrl, config)`. 4. **Garde langue** (CDC §3.7, adaptation runner documentée) : si `pageSemantics.lang` non vide ET ≠ `keyword.lang` (comparaison sur le sous-tag primaire) → la page passe `semantics: null` + `fetchOk: true` trafiqué en `false`?? NON — passe `semantics: null, fetchOk: false` et collecte `lang_mismatch:${position}` dans une liste TS locale (le crate la classera `missing` ; on ré-étiquette après, voir 6). 5. `buildSerpProfile(pageInputs, client, config)` — `client` = `payload.clientUrl` fetché/extrait (même chemin), sinon `null`. **La MÊME config aux deux appels wasm** (constraint globale). 6. Post-traitement TS du profil : appende les flags `lang_mismatch:${pos}` à `profile.coverage.flags` ; construis `StoredSemanticProfile` avec `analyzedAt: deps.now()`, `jobId`, `exports`. 7. Exports full-grain sur R2 (AVANT le merge DB — un profil persisté doit toujours pointer des exports existants) : - `pages.jsonl.gz` : 1 ligne JSON par page = `{ position, url, included, fetchOk, structure, archetype, lang }` ; - `terms.jsonl.gz` : 1 ligne par terme×page×zone = `{ term, position, zone, freq }` (zones du contrat : `title|h1|h2h3|slug|alt|anchor|meta|body`, n'émets que les freq > 0) ; - clés : `serp-semantics-results/${siteId}/${payload.keywordId}/${payload.country}/${jobId}/{pages,terms}.jsonl.gz`. - Si la ligne `keywords.semantic[country]` précédente existait, `deleteObject` best-effort ses anciens exports (design D4 : on ne garde que la dernière analyse). 8. Merge jsonb par-locale (constraint globale — SQL `coalesce(...) || jsonb_build_object(...)`). 9. `finally` : delete staging best-effort (toutes les clés uploadées), succès comme échec. - [ ] **Step 1: Tests d'abord.** Sur la DB de test (pattern `assertTestDb` — regarde comment `crawl-runner` est testé et copie le harnais). Fixtures : un site + un mot-clé FR seedés. Deps fake : `fetchHtml` renvoyant des HTML inline (dont une page dans une autre langue pour le test lang_mismatch), `putObject`/`deleteObject` enregistreurs, `now` fixe. Cas minimum : ```typescript test("pipeline complet : profil persisté par-locale + exports écrits + staging nettoyé", async () => { // 3 urls incluses (2 FR ok, 1 EN → lang_mismatch), 1 exclue (included:false), 1 fetch null const res = await runSerpSemantics("job-1", siteId, payload, deps); expect(res.ok).toBe(true); const row = await db.query.keywords.findFirst({ where: eq(keywords.id, kwId) }); const stored = row?.semantic?.FR; expect(stored?.jobId).toBe("job-1"); expect(stored?.coverage.excludedByUser).toEqual([4]); expect(stored?.coverage.missing).toContain(5); // fetch null expect(stored?.coverage.flags).toContain("lang_mismatch:3"); expect(stored?.exports.termsKey).toContain("job-1/terms.jsonl.gz"); expect(putKeys.some((k) => k.endsWith("terms.jsonl.gz"))).toBe(true); expect(deletedKeys.filter((k) => k.includes("staging")).length).toBeGreaterThan(0); }); test("merge par-locale : une analyse US ne détruit pas le profil FR existant", async () => { /* seed semantic.FR, run US, assert les deux clés présentes */ }); test("échec fetch total → ok:false, rien persisté, staging nettoyé", async () => { /* fetchHtml → null partout */ }); ``` - [ ] **Step 2: Vérifier l'échec.** — Run: `bun run test src/lib/semantic/analysis-runner.test.ts` (avec la DATABASE_URL de test selon le pattern repo). - [ ] **Step 3: Implémenter** selon la logique imposée. `terms.jsonl` : itère `pageInputs` avec `semantics`, puis `Object.entries(semantics.termFreqs)` × les 8 zones de `ZoneCounts`. - [ ] **Step 4: Vérifier** — tests PASS. - [ ] **Step 5: Commit** — `git commit -m "feat(semantic): runner DI du pipeline serp-semantics (fetch pool, staging R2, wasm, exports jsonl, merge par-locale)"` --- ### Task 4: Câblage service analytics — job + worker + garde anti-doublon **Files:** - Modify: `services/analytics/src/index.ts` (schéma zod du POST /jobs + garde 409) - Modify: `services/analytics/src/worker.ts` (entrée `HANDLERS`) - Create: `services/analytics/src/serp-semantics.ts` (adaptateur mince, mirror `site-crawl.ts`) - Modify: `src/lib/analytics/enqueue.ts` (shape du nouveau job) - Test: `services/analytics/test/serp-semantics.test.ts` (si le service a un harnais de test — regarde `services/analytics/test/` ; sinon teste l'adaptateur via le harnais des tests runner) **Interfaces:** - Consumes: `runSerpSemantics` (Task 3), `updateJobTotals`/`completeJob`/`failJob` (`services/analytics/src/queue.ts`), storage du service (`services/analytics/src/storage.ts` — réutilise ses primitives pour `putObject`/`deleteObject` ; ajoute `putObject` si seule la suppression existe, mirror du client S3 déjà en place). - Produces: type de job `serp-semantics` accepté par `POST /jobs` avec body zod : ```typescript const serpSemanticsJob = z.object({ type: z.literal("serp-semantics"), siteId: z.string().min(1), payload: z.object({ keywordId: z.string().min(1), country: z.string().length(2), urls: z.array(z.object({ position: z.number().int().min(1).max(10), url: z.string().url(), included: z.boolean() })).min(1).max(10), clientUrl: z.string().url().optional(), }), }); ``` - [ ] **Step 1: Test de la garde anti-doublon + dispatch** (selon le harnais existant du service ; a minima un test d'insertion : 2 enqueues même keywordId → le 2ᵉ reçoit 409/erreur tant que le 1ᵉʳ est `queued|running`. Requête : `analytics_jobs` where `type='serp-semantics'` AND `status IN ('queued','running')` AND `payload->>'keywordId' = $kw`). - [ ] **Step 2: Vérifier l'échec.** - [ ] **Step 3: Implémenter** : - `index.ts` : union zod étendue ; à l'insert, colonnes `siteId`, `type`, `payload` ; garde 409 (mirror de la garde site-crawl). - `serp-semantics.ts` (adaptateur) : câble `db` du service, `fetchSemanticHtml` (import depuis `src/lib/semantic/fetch.ts` — le service importe déjà du code `src/lib/**`, cf. `site-crawl.ts`), storage réel, `now: () => new Date().toISOString()`, `onProgress` → `updateJobTotals`. - `worker.ts` : `HANDLERS["serp-semantics"] = processSerpSemantics` ; succès → `completeJob`, échec → `failJob(error)`. - `enqueue.ts` : accepte le nouveau shape (union TS élargie, même POST). - [ ] **Step 4: Vérifier** — tests PASS + `bunx tsc --noEmit` clean (le service a son propre tsconfig ? vérifie comment il est typé-checké et applique la gate du repo). - [ ] **Step 5: Commit** — `git commit -m "feat(semantic): job serp-semantics dans le service analytics (zod, handler, garde anti-doublon)"` --- ### Task 5: Action Next de déclenchement + lecteur **Files:** - Create: `src/lib/semantic/actions.ts` - Test: `src/lib/semantic/actions.test.ts` (harnais DB de test, mirror des tests d'actions existants du domaine keywords) **Interfaces:** - Consumes: `assertKeywordInActiveOrg`, `enqueueAnalyticsJob`, `getAnalyticsJob` (polling existant, PAS de nouveau canal), schéma `keywords` (`serp`, `semantic`, `country`, `siteId`). - Produces (consommé par S4 UI) : ```typescript "use server"; // startSerpSemanticsAction(keywordId, opts?) : // 1. assertKeywordInActiveOrg(keywordId) → siteId // 2. hook no-op assertCanAnalyze(orgId) (D6 — commentaire « brancher quota/credits ici ») // 3. lit keywords.serp[country] → competitors (positions+urls). Pas de snapshot → {ok:false, error:"no_serp"} // 4. applique la curation : opts.excludedPositions → included:false ; clamp positions 1-10 // 5. enqueue { type:"serp-semantics", siteId, payload } → {ok:true, jobId} export async function startSerpSemanticsAction( keywordId: string, opts?: { excludedPositions?: number[]; clientUrl?: string }, ): Promise<{ ok: true; jobId: string } | { ok: false; error: string }>; // Lecture du profil stocké (RSC/UI) — tenancy incluse. export async function getSemanticProfileAction( keywordId: string, ): Promise; ``` - [ ] **Step 1: Tests** — no_serp (mot-clé sans snapshot), curation propagée (`excludedPositions:[2]` → l'URL pos 2 part `included:false`), clamp (snapshot avec pos 11+ → filtrées), tenancy (mot-clé d'une autre org → throw). Mock d'`enqueueAnalyticsJob` par DI ou spy selon le pattern du repo (attention mémoire projet : PAS de `mock.module` global — préfère l'injection). - [ ] **Step 2: Vérifier l'échec.** - [ ] **Step 3: Implémenter.** Si le pattern DI n'est pas naturel pour une server action, extraire la logique pure dans une fonction interne testable (`buildSerpSemanticsPayload(row, opts)`) et ne tester l'action que sur les chemins tenancy/no_serp. - [ ] **Step 4: Gates finales du plan** Run: `bun run test:all` → aucune régression (le lint `slugify.test.ts` pré-existant reste hors périmètre). Run: `bunx tsc --noEmit` → clean. Run: `cargo test -p semantic-core` → 62 (rien touché côté Rust). - [ ] **Step 5: Commit + push (PAS de merge dev)** ```bash git add -A && git commit -m "feat(semantic): startSerpSemanticsAction + lecteur de profil (curation, clamp, tenancy)" git push -u origin ytg-like/serp-pipeline ``` --- ## Self-review (fait à l'écriture) - Couverture design S3 : migration+lz4 ✔ (T1), job type+payload ✔ (T1/T4), runner DI fetch mobile/staging/wasm/exports/merge ✔ (T2/T3), garde anti-doublon ✔ (T4), action+curation+clamp+hook quota ✔ (T5). Parquet → JSON Lines : déviation documentée en Global Constraints (CDC §4.4 l'autorise). `lang_mismatch` : adaptation runner (exclusion scoring via missing + flag ré-étiqueté) documentée T3 — le statut intermédiaire crate reste un follow-up S5+. - Types cohérents : `SerpSemanticsPayload`/`StoredSemanticProfile` définis T1, consommés T3/T4/T5 sous les mêmes noms ; `RunnerDeps` défini T3, consommé T4. - Zéro polling nouveau : l'UI S4 réutilisera `getAnalyticsJob`.