# Design — Track analytique DuckDB/Hono (ETL + agrégation lourde)

> Spec issue du brainstorming du 2026-06-09. Premier sous-projet du "track analytique"
> différé dans `2026-06-09-data-to-content-flow-design.md` (§9).

## 1. Vision

Décharger le traitement des **métriques massives** (exports Semrush/Ahrefs/Mangools,
10k–100k+ lignes) hors du chemin synchrone actuel (rust-wasm en Server Action, limite
1 Mo). Un service **Bun + Hono** dédié laisse **DuckDB** faire tout le crunch
(parse/filtre/dédoublonne/normalise) et n'upsert que le **résultat curé** dans Postgres
via Drizzle. **Postgres reste la source de vérité** ; DuckDB est le moteur de calcul.

## 2. Périmètre

**Dans cette spec** — le pipeline ETL+agrégation v1, de bout en bout :
upload CSV → object storage → queue → DuckDB → upsert Drizzle → statut.

**Hors périmètre (tracks suivants)** : consensus/best-metric cross-source calculé ;
dashboards analytiques en lecture DuckDB ; workers Rust ; migration SurrealDB.

## 3. Architecture

```
Next (UI upload, scopé site)
  (1) PUT CSV  ─────────────────────────────►  Object storage (MinIO/S3, Dokploy)
  (2) enqueue  ─────────────────────────────►  analytics_jobs {status: queued}
                                                      │ (Postgres partagé)
Service Bun + Hono (process séparé, Dokploy)          │
  (3) worker poll queued ◄────────────────────────────┘
  (4) DuckDB :memory: + httpfs → read_csv('s3://…/objectKey')
  (5) SQL: normalise colonnes (par source) → dédoublonne → matche keywords (term+lang+country)
  (6) result set curé (~N lignes) → upsert Drizzle (keywords.metrics[source] + nouveaux keywords)
  (7) analytics_jobs.status = done|failed (+ totals/error)
Next  ◄── lit le statut (même Postgres) ── toast/notif
```

## 4. Composants (unités isolées, interfaces claires)

- **`services/analytics/`** — app Bun + Hono autonome (`package.json` propre, micro-module).
  Importe le **schéma Drizzle existant** (`src/lib/db/domain-schema`) + un client `db` =
  source de vérité unique du schéma. Expose une route d'enqueue typée (`zValidator` + export
  `AppType` pour `hc<AppType>`), et fait tourner une **boucle worker** qui consomme la table.
- **DuckDB processor** (`services/analytics/src/processor.ts`) — par job : DB DuckDB
  **éphémère** (`:memory:`), extension `httpfs/s3` (secret bucket via env), `read_csv` depuis
  l'object storage, SQL de normalisation/dédup/match, retourne `{ rows: CuratedMetric[], totals }`.
  Cleanup (DETACH/close) en `finally`.
- **Normaliseurs par source** (`services/analytics/src/normalizers/`) — mapping colonnes
  Semrush/Ahrefs/Mangools → `SourceMetrics` commun, exprimé en SQL DuckDB (un fichier .sql ou
  builder par source). Le rust `import-core` reste pour le petit chemin synchrone existant.
- **Upsert Drizzle** (`services/analytics/src/upsert.ts`) — pour chaque `CuratedMetric` :
  set `keywords.metrics[source] = metrics` (merge jsonb) sur le keyword matché ; **crée** les
  keywords absents (`status: "idea"`, `primarySource = source`, locale du site). Scopé `siteId`.
- **`analytics_jobs`** (nouvelle table Postgres) — la queue + le statut.
- **Next : UI upload + statut** — un écran/section (réutilise le pattern d'upload existant)
  qui PUT le CSV vers le bucket, insère le job, et affiche la progression (lecture du job).

## 5. Modèle de données — `analytics_jobs`

```ts
export const analyticsJobs = pgTable("analytics_jobs", {
  id: text("id").primaryKey(),                 // uuid
  siteId: text("site_id").notNull().references(() => sites.id, { onDelete: "cascade" }),
  source: text("source").$type<KeywordSource>().notNull(),  // semrush | ahrefs | mangools
  objectKey: text("object_key").notNull(),     // clé bucket (dérivée serveur, jamais brute client)
  status: text("status").$type<"queued"|"running"|"done"|"failed">().notNull().default("queued"),
  totals: jsonb("totals").$type<{ read?: number; matched?: number; created?: number; upserted?: number }>().notNull().default({}),
  error: text("error"),
  createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
  updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
});
```
Migration via `db:generate` + `db:migrate` (jamais `db:push`).

**`CuratedMetric`** (interface partagée service↔processor) :
```ts
interface CuratedMetric {
  term: string; lang: string; country: string;
  source: KeywordSource;
  metrics: SourceMetrics; // {volume, kd, cpc, growthPct, competitiveDensity?, numberOfResults?, updatedAt}
}
```

## 6. Flux détaillé

1. **Upload** — Next : l'utilisateur (site actif) dépose un export ; Next PUT le fichier vers
   `s3://exports/{siteId}/{uuid}.csv` (clé dérivée serveur), puis insère `analytics_jobs`
   (`source` choisi, `objectKey`, `status: queued`). Retour immédiat (job id).
2. **Pick** — le worker du service poll `analytics_jobs WHERE status='queued'` (FOR UPDATE SKIP
   LOCKED), passe `running`.
3. **Crunch DuckDB** — `read_csv` depuis le bucket → SQL normalise (colonnes du `source`) →
   dédoublonne (1 ligne/keyword, dernière/meilleure) → produit `CuratedMetric[]`.
4. **Match + upsert (Drizzle)** — matche par (term, lang, country) sur le site ; met à jour
   `metrics[source]` des existants, **crée** les nouveaux ; renseigne `totals`.
5. **Statut + cleanup** — `status: done` (+ totals) → **supprime l'objet du bucket** (pas de
   rétention ; les données curées sont désormais en Postgres). `failed` (+ error) → l'objet est
   **conservé** pour permettre un retry. Next lit le statut et notifie (toast).

## 7. Robustesse, sécurité, tests

- **Idempotence** : re-uploader le même export = même résultat (upsert, pas d'append). Le
  fichier source est **supprimé du bucket après succès** (pas de rétention) ; un job `failed`
  conserve l'objet pour retry. Worker ne crash pas sur échec (`try/finally`, `status: failed`).
- **Sécurité** (leçons du proto Hono) : `objectKey` **dérivé/validé** serveur (jamais de chemin
  client brut → pas de LFI) ; creds bucket + Postgres via **env** (DuckDB `CREATE SECRET`) ;
  requêtes DuckDB **paramétrées** (zéro interpolation) ; enqueue validé `zValidator`.
- **Concurrence** : une connexion DuckDB **par job** ; `SKIP LOCKED` évite le double-pick.
- **Tenancy** : job scopé `siteId` ; l'upsert n'écrit que des keywords de ce site.
- **Tests** : normaliseurs par source (SQL DuckDB sur petit CSV fixture → `SourceMetrics`
  attendu) ; matching (term+locale) ; round-trip job (enqueue → process → upsert vérifié en DB) ;
  idempotence (2 runs = 1 résultat). DuckDB tourne en CI sans réseau (fixture CSV local).

## 8. Décisions actées
1. Queue = table Postgres `analytics_jobs` (pas de Redis ; statut lu directement par l'app).
2. Les nouveaux keywords de l'export sont **créés** (`status: "idea"`, `primarySource = source`),
   en plus de la mise à jour des existants.
3. Agrégation v1 = **par source** (normalise/dédup/upsert `metrics[source]`) ; consensus
   cross-source calculé = ajout ultérieur.
4. DuckDB **éphémère** par job ; résultat curé dans Postgres.
5. Service dans le **même repo** (`services/analytics/`), partage le schéma Drizzle.
6. Écriture PG via **upsert Drizzle** (pas l'extension postgres de DuckDB) — l'agrégat est petit.
7. **Pas de rétention** : l'objet uploadé est supprimé du bucket après import réussi (source de
   vérité = Postgres) ; conservé seulement si le job échoue (retry).

## 9. Différé (tracks suivants)
Consensus/best-metric cross-source · dashboards lecture DuckDB · workers Rust · SurrealDB.