# DuckDB/Hono Analytics Pipeline — 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:** A standalone Bun + Hono service that ingests big keyword CSV exports (Semrush/Mangools) via DuckDB and upserts curated per-source metrics into Postgres through Drizzle, driven by a Postgres-backed job queue.

**Architecture:** Next uploads a CSV to object storage and inserts an `analytics_jobs` row; a separate Bun service polls the queue, lets DuckDB `read_csv` + normalize + dedup the file (heavy crunch stays in DuckDB), then upserts the small curated result via its own Drizzle client. Postgres stays the source of truth. The uploaded file is deleted on success.

**Tech Stack:** Bun, Hono (+ @hono/zod-validator), DuckDB (`@duckdb/node-api`), Drizzle/postgres-js, Bun native S3 (`Bun.s3`), zod, biome.

**Spec:** `docs/superpowers/specs/2026-06-09-duckdb-analytics-pipeline-design.md`

---

## Conventions
- Tests: `bun test` (the service has its own `bun test`); DB round-trip checks run against the dev DB with `bun --env-file=.env.local --conditions=react-server run <script>` (the `db:migrate:test` suite is blocked by an unrelated `.env.test` issue). Dev site for manual checks: `ac5b0f4a-c760-459d-80e2-43786951381f`.
- `npx tsc --noEmit` + `npx @biomejs/biome check <files>` must be clean before each commit.
- Never `db:push`; migrations via `db:generate` + `db:migrate` (loads `.env.local` via the config fix already in place).
- The service must NOT import `@/lib/db/client` (it pulls in Next-coupled `@/env`). It builds its own postgres-js client but **imports the schema** `@/lib/db/domain-schema` and types `@/lib/keywords/types` (pure defs) via tsconfig `paths`.

## File Structure
**Modify (main app)**
- `src/lib/db/domain-schema.ts` — add the `analyticsJobs` table.

**Create (service)**
- `services/analytics/package.json`, `tsconfig.json`
- `services/analytics/src/types.ts` — `CanonicalKeyword`, `CuratedResult`.
- `services/analytics/src/db.ts` — own Drizzle client (no `@/env`).
- `services/analytics/src/storage.ts` — object access (Bun S3 in prod, local path in tests) + delete.
- `services/analytics/src/processor.ts` — DuckDB: read_csv → normalize (per source) → dedup → `CanonicalKeyword[]`.
- `services/analytics/src/upsert.ts` — Drizzle upsert (copied merge SQL).
- `services/analytics/src/queue.ts` — claim (SKIP LOCKED) / complete / fail.
- `services/analytics/src/worker.ts` — process one job end-to-end.
- `services/analytics/src/index.ts` — Hono app (enqueue route, `AppType`) + worker loop.
- `services/analytics/test/*.test.ts`, `services/analytics/test/fixtures/*.csv`

---

## Task 1: `analytics_jobs` table + migration

**Files:** Modify `src/lib/db/domain-schema.ts`

- [ ] **Step 1: Add the table.** After the `keywordTargets` table, insert (the imports `pgTable,text,jsonb,timestamp` already exist; `KeywordSource` is already imported):
```ts
/** Async ETL jobs for the analytics service (queue + status). */
export const analyticsJobs = pgTable("analytics_jobs", {
  id: text("id").primaryKey(),
  siteId: text("site_id")
    .notNull()
    .references(() => sites.id, { onDelete: "cascade" }),
  source: text("source").$type<KeywordSource>().notNull(),
  objectKey: text("object_key").notNull(),
  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(),
});
```

- [ ] **Step 2: Generate + apply migration.**
```bash
bun run db:generate           # creates drizzle/00NN_*.sql with analytics_jobs
bun run src/lib/db/migrate.ts # (loads .env.local via drizzle config); or: bun --env-file=.env.local --conditions=react-server run src/lib/db/migrate.ts
```
- [ ] **Step 3: Verify table exists.**
```bash
psql "$(grep -hE '^DATABASE_URL' .env.local | head -1 | cut -d= -f2- | tr -d '"')" -c "\d analytics_jobs"
```
Expected: columns `id, site_id, source, object_key, status, totals, error, created_at, updated_at`.
- [ ] **Step 4: tsc + commit.**
```bash
npx tsc --noEmit && npx @biomejs/biome check src/lib/db/domain-schema.ts
git add src/lib/db/domain-schema.ts drizzle/
git commit -m "feat(db): analytics_jobs table (ETL queue + status)"
```

---

## Task 2: Service scaffold + DB client

**Files:** Create `services/analytics/{package.json,tsconfig.json,src/db.ts,src/types.ts}`

- [ ] **Step 1: package.json**
```json
{
  "name": "@seo-insight/analytics",
  "private": true,
  "type": "module",
  "scripts": {
    "dev": "bun run src/index.ts",
    "test": "bun test",
    "check": "biome check src && tsc --noEmit"
  },
  "dependencies": {
    "hono": "^4.6.0",
    "@hono/zod-validator": "^0.4.0",
    "zod": "^3.23.0",
    "@duckdb/node-api": "^1.1.3",
    "drizzle-orm": "^0.45.2",
    "postgres": "^3.4.9"
  }
}
```
- [ ] **Step 2: tsconfig.json** (maps `@/*` to the app src so we can import the schema/types):
```json
{
  "compilerOptions": {
    "target": "ESNext", "module": "ESNext", "moduleResolution": "bundler",
    "strict": true, "skipLibCheck": true, "noEmit": true, "types": ["bun"],
    "baseUrl": ".", "paths": { "@/*": ["../../src/*"] }
  },
  "include": ["src", "test"]
}
```
- [ ] **Step 3: install + smoke-test DuckDB on Bun** (verify the native dep loads — if `@duckdb/node-api` fails under Bun, fall back to `duckdb-async` and adjust the processor in Task 5):
```bash
cd services/analytics && bun install
bun -e "import {DuckDBInstance} from '@duckdb/node-api'; const i=await DuckDBInstance.create(':memory:'); const c=await i.connect(); const r=await c.run('select 42 as x'); console.log('DUCKDB_OK', (await r.getRows())[0][0]);"
```
Expected: `DUCKDB_OK 42`. If it errors, switch the dep to `duckdb-async` and note it.
- [ ] **Step 4: src/types.ts**
```ts
import type { KeywordSource, SearchIntent, SourceMetrics } from "@/lib/keywords/types";

export interface CanonicalKeyword {
  slug: string;
  term: string;
  lang: string;
  country: string;
  intent: SearchIntent[];
  serpFeatures: string[];
  primarySource: KeywordSource;
  metrics: SourceMetrics;
}
export interface CuratedResult {
  rows: CanonicalKeyword[];
  totals: { read: number };
}
```
- [ ] **Step 5: src/db.ts** (own client; reads DATABASE_URL from env, NOT @/env):
```ts
import { drizzle } from "drizzle-orm/postgres-js";
import postgres from "postgres";
import * as domainSchema from "@/lib/db/domain-schema";

const url = process.env.DATABASE_URL;
if (!url) throw new Error("DATABASE_URL is required");
const client = postgres(url);
export const db = drizzle(client, { schema: domainSchema });
export { client };
```
- [ ] **Step 6: commit**
```bash
npx tsc -p services/analytics/tsconfig.json --noEmit && npx @biomejs/biome check services/analytics/src
git add services/analytics
git commit -m "feat(analytics): Bun service scaffold + drizzle client"
```

---

## Task 3: Storage (object access + delete)

**Files:** Create `services/analytics/src/storage.ts`, Test `services/analytics/test/storage.test.ts`

Object access abstracted so tests use a **local file path** and prod uses S3/MinIO (Bun native `Bun.s3`). `readCsvLocation(key)` returns a string DuckDB can `read_csv()` (a local path in tests, or a presigned/`s3://` URL with the duckdb httpfs secret in prod). For v1 keep it simple: tests pass an absolute local path as the `objectKey`; prod wires Bun S3.

- [ ] **Step 1: failing test** `test/storage.test.ts`:
```ts
import { expect, test } from "bun:test";
import { resolveReadable, deleteObject } from "../src/storage";

test("resolveReadable returns a local path as-is when STORAGE_DRIVER=local", () => {
  process.env.STORAGE_DRIVER = "local";
  expect(resolveReadable("/tmp/x.csv")).toBe("/tmp/x.csv");
});
test("deleteObject is a no-op for local driver and does not throw", async () => {
  process.env.STORAGE_DRIVER = "local";
  await deleteObject("/tmp/x.csv"); // must resolve
  expect(true).toBe(true);
});
```
- [ ] **Step 2: run → fail** (`cd services/analytics && bun test test/storage.test.ts`): module not found.
- [ ] **Step 3: implement** `src/storage.ts`:
```ts
const driver = () => process.env.STORAGE_DRIVER ?? "s3";

/** A location string DuckDB read_csv() can open. local → the path; s3 → an s3:// URL. */
export function resolveReadable(objectKey: string): string {
  if (driver() === "local") return objectKey;
  const bucket = process.env.S3_BUCKET ?? "exports";
  return `s3://${bucket}/${objectKey}`;
}

/** Delete the uploaded object after a successful import (no retention). */
export async function deleteObject(objectKey: string): Promise<void> {
  if (driver() === "local") return; // tests / fixtures: never delete
  const bucket = process.env.S3_BUCKET ?? "exports";
  await Bun.s3.file(objectKey, { bucket }).delete();
}
```
- [ ] **Step 4: run → pass.**
- [ ] **Step 5: commit** `git commit -m "feat(analytics): storage access + delete-after-import"`

---

## Task 4: DuckDB processor (read_csv → normalize → dedup)

**Files:** Create `services/analytics/src/processor.ts`, Test `services/analytics/test/processor.test.ts`, fixtures `test/fixtures/{semrush,mangools}.csv`

The processor replicates the Rust column mapping in DuckDB SQL. Source detection by header; per-source SELECT mapping CSV columns → `CanonicalKeyword`. Slug = lowercase term, non-alphanumerics → single `-`, trimmed. Dedup: one row per slug (highest volume). Trend is deferred (`[]`) for v1.

Column mapping (from the Rust parser):
- **Semrush** headers: `Keyword, Intent, Volume, Trend, Keyword Difficulty, CPC (USD), Competitive Density, SERP Features, Number of Results`.
- **Mangools** headers: `Keyword, Avg. Search Volume (Last Known Values), Location, CPC/USD, Keyword Difficulty, Keyword interest growth (%), Search Intent, Serp Features`.

- [ ] **Step 1: fixtures.** `test/fixtures/semrush.csv`:
```
Keyword,Intent,Volume,Trend,Keyword Difficulty,CPC (USD),Competitive Density,SERP Features,Number of Results
seo audit,Informational,1000,"0.1,0.2",45,2.50,0.30,"Featured Snippet",1200000
seo audit,Informational,1000,"0.1,0.2",45,2.50,0.30,"Featured Snippet",1200000
link building,Commercial,800,"0.3",60,4.10,0.55,"",900000
```
`test/fixtures/mangools.csv`:
```
Keyword,Avg. Search Volume (Last Known Values),Location,CPC/USD,Keyword Difficulty,Keyword interest growth (%),Search Intent,Serp Features
netlinking,500,France,1.20,35,12.5,Commercial,"Sitelinks"
```
- [ ] **Step 2: failing test** `test/processor.test.ts`:
```ts
import { expect, test } from "bun:test";
import { join } from "node:path";
import { processCsv, detectSource } from "../src/processor";

const fx = (n: string) => join(import.meta.dir, "fixtures", n);

test("detectSource by header", async () => {
  expect(await detectSource(fx("semrush.csv"))).toBe("semrush");
  expect(await detectSource(fx("mangools.csv"))).toBe("mangools");
});

test("semrush: maps columns, dedups by slug, slugs term", async () => {
  const { rows, totals } = await processCsv(fx("semrush.csv"), "semrush");
  expect(totals.read).toBe(3);
  expect(rows.length).toBe(2); // duplicate "seo audit" collapsed
  const audit = rows.find((r) => r.slug === "seo-audit")!;
  expect(audit.term).toBe("seo audit");
  expect(audit.metrics.volume).toBe(1000);
  expect(audit.metrics.kd).toBe(45);
  expect(audit.metrics.cpc).toBe(2.5);
  expect(audit.metrics.competitiveDensity).toBe(0.3);
  expect(audit.metrics.numberOfResults).toBe(1200000);
  expect(audit.primarySource).toBe("semrush");
  expect(audit.country).toBe("US");
});

test("mangools: growthPct + country from Location", async () => {
  const { rows } = await processCsv(fx("mangools.csv"), "mangools");
  const r = rows[0];
  expect(r.slug).toBe("netlinking");
  expect(r.metrics.volume).toBe(500);
  expect(r.metrics.growthPct).toBe(12.5);
  expect(r.metrics.competitiveDensity).toBeNull();
});
```
- [ ] **Step 3: implement** `src/processor.ts` (uses `@duckdb/node-api`; adapt to `duckdb-async` if Task 2 fell back). Key points: read header for detection, run a per-source SQL SELECT with `read_csv(?, header=true, all_varchar=true)`, build `CanonicalKeyword[]` in TS from the result rows, dedup by slug keeping max volume, set `updatedAt` to `new Date().toISOString()`.
```ts
import { DuckDBInstance } from "@duckdb/node-api";
import type { KeywordSource, SearchIntent } from "@/lib/keywords/types";
import type { CanonicalKeyword, CuratedResult } from "./types";
import { resolveReadable } from "./storage";

function slugify(s: string): string {
  return s.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-+|-+$/g, "");
}
const intents: Record<string, SearchIntent> = {
  informational: "informational", commercial: "commercial",
  transactional: "transactional", navigational: "navigational",
};
function splitIntents(s: string): SearchIntent[] {
  return s.split(/[,;|]/).map((x) => intents[x.trim().toLowerCase()]).filter(Boolean) as SearchIntent[];
}
function splitSerp(s: string): string[] {
  return s.split(/[,;|]/).map((x) => x.trim()).filter(Boolean);
}
const num = (v: unknown): number => { const n = Number(String(v ?? "").replace(/[^0-9.\-]/g, "")); return Number.isFinite(n) ? n : 0; };
const numOrNull = (v: unknown): number | null => { const s = String(v ?? "").trim(); if (!s) return null; const n = Number(s.replace(/[^0-9.\-]/g, "")); return Number.isFinite(n) ? n : null; };

export async function detectSource(loc: string): Promise<KeywordSource | "unknown"> {
  const inst = await DuckDBInstance.create(":memory:");
  const c = await inst.connect();
  const reader = await c.runAndReadAll(
    "select * from read_csv(?, header=true, all_varchar=true) limit 0",
    [resolveReadable(loc)],
  );
  const cols = reader.columnNames();
  if (cols.includes("Avg. Search Volume (Last Known Values)")) return "mangools";
  if (cols.includes("Competitive Density") && cols.includes("Number of Results")) return "semrush";
  return "unknown";
}

export async function processCsv(loc: string, source: KeywordSource): Promise<CuratedResult> {
  const inst = await DuckDBInstance.create(":memory:");
  const c = await inst.connect();
  const readable = resolveReadable(loc);
  const cols = source === "semrush"
    ? `"Keyword","Volume","Keyword Difficulty","CPC (USD)","Competitive Density","Number of Results","Intent","SERP Features"`
    : `"Keyword","Avg. Search Volume (Last Known Values)","Keyword Difficulty","CPC/USD","Keyword interest growth (%)","Search Intent","Serp Features","Location"`;
  const reader = await c.runAndReadAll(
    `select ${cols} from read_csv(?, header=true, all_varchar=true)`,
    [readable],
  );
  const recs = reader.getRowObjects() as Record<string, string>[];
  const updatedAt = new Date().toISOString();
  const bySlug = new Map<string, CanonicalKeyword>();
  for (const r of recs) {
    const term = (r["Keyword"] ?? "").trim();
    if (!term) continue;
    const slug = slugify(term);
    if (!slug) continue;
    const k: CanonicalKeyword = source === "semrush"
      ? {
          slug, term, lang: "en", country: "US",
          intent: splitIntents(r["Intent"] ?? ""), serpFeatures: splitSerp(r["SERP Features"] ?? ""),
          primarySource: "semrush",
          metrics: { volume: num(r["Volume"]), kd: num(r["Keyword Difficulty"]), cpc: num(r["CPC (USD)"]),
            growthPct: null, competitiveDensity: numOrNull(r["Competitive Density"]),
            numberOfResults: numOrNull(r["Number of Results"]), updatedAt },
        }
      : {
          slug, term, lang: "en", country: countryFromLocation(r["Location"] ?? ""),
          intent: splitIntents(r["Search Intent"] ?? ""), serpFeatures: splitSerp(r["Serp Features"] ?? ""),
          primarySource: "mangools",
          metrics: { volume: num(r["Avg. Search Volume (Last Known Values)"]), kd: num(r["Keyword Difficulty"]),
            cpc: num(r["CPC/USD"]), growthPct: numOrNull(r["Keyword interest growth (%)"]),
            competitiveDensity: null, numberOfResults: null, updatedAt },
        };
    const prev = bySlug.get(slug);
    if (!prev || k.metrics.volume > prev.metrics.volume) bySlug.set(slug, k);
  }
  return { rows: [...bySlug.values()], totals: { read: recs.length } };
}

function countryFromLocation(loc: string): string {
  const m: Record<string, string> = { france: "FR", "united states": "US", "united kingdom": "GB", canada: "CA", australia: "AU" };
  return m[loc.trim().toLowerCase()] ?? "US";
}
```
> The `@duckdb/node-api` method names (`runAndReadAll`, `getRowObjects`, `columnNames`) are from its current API; if they differ in the installed version, adapt to the equivalent read methods (do not change the SQL).
- [ ] **Step 4: run → pass** (`bun test test/processor.test.ts`). Fix until green.
- [ ] **Step 5: commit** `git commit -m "feat(analytics): DuckDB CSV processor (semrush/mangools → canonical)"`

---

## Task 5: Drizzle upsert

**Files:** Create `services/analytics/src/upsert.ts`, Test `services/analytics/test/upsert.test.ts`

Reuses the exact merge semantics from `src/lib/keywords/import.ts` (metrics jsonb merge, sources union, theme/status preserved) but against the service's own `db`.

- [ ] **Step 1: failing round-trip test** `test/upsert.test.ts` (DB-gated; uses the dev DB + a temp site or the known site):
```ts
import { expect, test } from "bun:test";
import { and, eq } from "drizzle-orm";
import { db } from "../src/db";
import { keywords } from "@/lib/db/domain-schema";
import { upsertCanonical } from "../src/upsert";
import type { CanonicalKeyword } from "../src/types";

const SITE = "ac5b0f4a-c760-459d-80e2-43786951381f";
test("upsertCanonical creates new + merges metrics per source", async () => {
  const slug = `zzqa-${Date.now()}`;
  const row: CanonicalKeyword = { slug, term: "zz qa", lang: "en", country: "US",
    intent: ["commercial"], serpFeatures: [], primarySource: "semrush",
    metrics: { volume: 10, kd: 5, cpc: 1, growthPct: null, competitiveDensity: 0.2, numberOfResults: 100, updatedAt: new Date().toISOString() } };
  const r = await upsertCanonical(SITE, [row]);
  expect(r.created + r.upserted).toBeGreaterThan(0);
  const id = `${SITE}-${slug}`;
  const k = (await db.select().from(keywords).where(eq(keywords.id, id)))[0];
  expect(k.metrics.semrush?.volume).toBe(10);
  // second source merges, not overwrites
  await upsertCanonical(SITE, [{ ...row, primarySource: "mangools", metrics: { ...row.metrics, volume: 20 } }]);
  const k2 = (await db.select().from(keywords).where(eq(keywords.id, id)))[0];
  expect(k2.metrics.semrush?.volume).toBe(10);
  expect(k2.metrics.mangools?.volume).toBe(20);
  await db.delete(keywords).where(eq(keywords.id, id));
});
```
- [ ] **Step 2: run → fail** (`bun --env-file=../../.env.local --conditions=react-server test test/upsert.test.ts` from `services/analytics`, or run via the app's test env). Import error.
- [ ] **Step 3: implement** `src/upsert.ts`:
```ts
import { sql } from "drizzle-orm";
import { keywords } from "@/lib/db/domain-schema";
import { db } from "./db";
import type { CanonicalKeyword } from "./types";

export async function upsertCanonical(
  siteId: string,
  rows: CanonicalKeyword[],
): Promise<{ created: number; upserted: number }> {
  if (rows.length === 0) return { created: 0, upserted: 0 };
  const values = rows.map((k) => ({
    id: `${siteId}-${k.slug}`,
    siteId, term: k.term, lang: k.lang, country: k.country, theme: null,
    status: "idea" as const, intent: k.intent,
    primarySource: k.primarySource, sources: [k.primarySource],
    metrics: { [k.primarySource]: k.metrics },
    trend: [], serpFeatures: k.serpFeatures,
  }));
  await db.insert(keywords).values(values).onConflictDoUpdate({
    target: keywords.id,
    set: {
      term: sql`excluded.term`, lang: sql`excluded.lang`, country: sql`excluded.country`,
      intent: sql`excluded.intent`, primarySource: sql`excluded.primary_source`,
      sources: sql`(select array_agg(distinct s) from unnest(${keywords.sources} || excluded.sources) as s)`,
      metrics: sql`${keywords.metrics} || excluded.metrics`,
      serpFeatures: sql`excluded.serp_features`,
    },
  });
  return { created: 0, upserted: values.length }; // counts refined in the worker via pre-check if needed
}
```
- [ ] **Step 4: run → pass.**
- [ ] **Step 5: commit** `git commit -m "feat(analytics): canonical upsert (metrics merge, sources union)"`

---

## Task 6: Queue (claim / complete / fail)

**Files:** Create `services/analytics/src/queue.ts`, Test `services/analytics/test/queue.test.ts`

- [ ] **Step 1: failing round-trip test** `test/queue.test.ts` (insert a job, claim it, complete it; assert status transitions). Use the `analyticsJobs` table + `db`. (Full test code: insert `{id, siteId:SITE, source:"semrush", objectKey:"x", status:"queued"}`, call `claimNextJob()` → status running, `completeJob(id,totals)` → done; cleanup delete.)
- [ ] **Step 2: run → fail.**
- [ ] **Step 3: implement** `src/queue.ts`:
```ts
import { eq, sql } from "drizzle-orm";
import { analyticsJobs } from "@/lib/db/domain-schema";
import { db } from "./db";

export type Job = typeof analyticsJobs.$inferSelect;

/** Atomically claim one queued job (SKIP LOCKED) and mark it running. */
export async function claimNextJob(): Promise<Job | null> {
  const rows = await db.execute(sql`
    update analytics_jobs set status='running', updated_at=now()
    where id = (select id from analytics_jobs where status='queued'
                order by created_at asc for update skip locked limit 1)
    returning *`);
  const r = (rows as unknown as { rows?: Job[] }).rows ?? (rows as unknown as Job[]);
  return r[0] ?? null;
}
export async function completeJob(id: string, totals: Job["totals"]): Promise<void> {
  await db.update(analyticsJobs).set({ status: "done", totals, updatedAt: new Date() }).where(eq(analyticsJobs.id, id));
}
export async function failJob(id: string, error: string): Promise<void> {
  await db.update(analyticsJobs).set({ status: "failed", error, updatedAt: new Date() }).where(eq(analyticsJobs.id, id));
}
```
> Verify the postgres-js `db.execute` return shape for the RETURNING rows and adapt the unwrap. If cleaner, express the claim with drizzle's query builder + a raw `for update skip locked` subquery.
- [ ] **Step 4: run → pass.** **Step 5: commit** `git commit -m "feat(analytics): postgres job queue (claim/complete/fail)"`

---

## Task 7: Worker (process one job end-to-end)

**Files:** Create `services/analytics/src/worker.ts`, Test `services/analytics/test/worker.test.ts`

- [ ] **Step 1: failing test** — insert a job whose `objectKey` is the local semrush fixture (`STORAGE_DRIVER=local`), run `processJob(job)`, assert: keywords created for the fixture terms, job `done`, object not deleted (local driver). Cleanup.
- [ ] **Step 2: run → fail.**
- [ ] **Step 3: implement** `src/worker.ts`:
```ts
import { deleteObject } from "./storage";
import { processCsv } from "./processor";
import { completeJob, failJob, type Job } from "./queue";
import { upsertCanonical } from "./upsert";

export async function processJob(job: Job): Promise<void> {
  try {
    const { rows, totals } = await processCsv(job.objectKey, job.source);
    const up = await upsertCanonical(job.siteId, rows);
    await completeJob(job.id, { read: totals.read, upserted: up.upserted });
    await deleteObject(job.objectKey); // no retention on success
  } catch (e) {
    await failJob(job.id, (e as Error).message); // object retained for retry
  }
}
```
- [ ] **Step 4: run → pass.** **Step 5: commit** `git commit -m "feat(analytics): worker — process job (duckdb → upsert → cleanup)"`

---

## Task 8: Hono app — enqueue route + worker loop

**Files:** Create `services/analytics/src/index.ts`, Test `services/analytics/test/enqueue.test.ts`

- [ ] **Step 1: failing test** — `app.request('/jobs', { method:'POST', body: JSON.stringify({siteId, source, objectKey}) })` returns 200 + `{ id }`, and a row exists in `analytics_jobs` with status `queued`. (Use Hono's `app.request`.)
- [ ] **Step 2: run → fail.**
- [ ] **Step 3: implement** `src/index.ts`:
```ts
import { randomUUID } from "node:crypto";
import { serve } from "bun";
import { Hono } from "hono";
import { zValidator } from "@hono/zod-validator";
import { z } from "zod";
import { analyticsJobs } from "@/lib/db/domain-schema";
import { db } from "./db";
import { claimNextJob } from "./queue";
import { processJob } from "./worker";

const Enqueue = z.object({
  siteId: z.string().min(1),
  source: z.enum(["semrush", "mangools", "ahrefs"]),
  objectKey: z.string().min(1),
});

const app = new Hono().post("/jobs", zValidator("json", Enqueue), async (c) => {
  const { siteId, source, objectKey } = c.req.valid("json");
  const id = randomUUID();
  await db.insert(analyticsJobs).values({ id, siteId, source, objectKey, status: "queued" });
  return c.json({ id });
});

export type AppType = typeof app;
export default app;

// Worker loop (only when run as the entrypoint, not when imported in tests).
if (import.meta.main) {
  serve({ port: Number(process.env.PORT ?? 8787), fetch: app.fetch });
  const tick = async (): Promise<void> => {
    let processed = false;
    try {
      const job = await claimNextJob();
      if (job) { await processJob(job); processed = true; }
    } catch (e) { console.error("[worker]", e); }
    setTimeout(tick, processed ? 0 : 2000); // drain fast, idle-poll every 2s
  };
  void tick();
}
```
- [ ] **Step 4: run → pass** (`bun test test/enqueue.test.ts`). **Step 5:** `cd services/analytics && bun run check`. **Step 6: commit** `git commit -m "feat(analytics): Hono enqueue route + worker loop (AppType for RPC)"`

---

## Task 9: Next integration — upload + enqueue + status

**Files:** Create `src/lib/analytics/actions.ts` (or a route handler) + a small UI entry in the keywords Import area; Modify nothing in the service.

- [ ] **Step 1:** Server action / route handler `uploadAndEnqueue(siteId, source, file)`: PUT the file to S3 under a **server-derived** key `exports/${siteId}/${randomUUID()}.csv` (Bun S3 / the app's storage util), then POST to the analytics service `/jobs` (typed via `hc<AppType>` from `@seo-insight/analytics`, base URL from `env.ANALYTICS_URL`) with `{ siteId, source, objectKey }`. Return the job id. (Guard `file.size`, validate `source`, scope `siteId` to the active site.)
- [ ] **Step 2:** A status reader `getAnalyticsJob(id)` (server query on `analyticsJobs`, scoped to the active org's sites) for the UI to poll; show a toast on `done`/`failed`.
- [ ] **Step 3:** Minimal UI: in the keywords import surface, an "Import lourd (gros export)" option that calls `uploadAndEnqueue` and polls status. Reuse existing import dialog patterns; do NOT exceed the 1MB Server Action limit (the file goes to S3 via a route handler / presigned PUT, not through a Server Action body).
- [ ] **Step 4:** `npx tsc --noEmit` + biome clean. **Step 5: commit** `git commit -m "feat(keywords): heavy CSV import via the analytics service"`

---

## Task 10: End-to-end verification + docs

- [ ] **Step 1:** Service tests green: `cd services/analytics && bun test` (processor + storage + queue; DB-gated upsert/worker/enqueue via `bun --env-file=../../.env.local --conditions=react-server test`).
- [ ] **Step 2:** Manual round-trip on the dev DB: insert a job pointing at a local fixture (`STORAGE_DRIVER=local`), run the worker once, confirm keywords upserted + job `done` (round-trip script). Clean up the inserted keywords.
- [ ] **Step 3:** `npx tsc --noEmit` (root) clean.
- [ ] **Step 4:** Document the service in `docs/project/overview.md` (the `services/analytics/` micro-module) + a short `services/analytics/README.md` (env: `DATABASE_URL`, `STORAGE_DRIVER`, `S3_*`, `PORT`). Commit.

---

## Notes / risks
- **DuckDB on Bun:** Task 2 Step 3 smoke-tests `@duckdb/node-api`. If it doesn't load under Bun, fall back to `duckdb-async` and adjust the processor's read calls (SQL unchanged).
- **`db.execute` RETURNING shape** (Task 6) and `@duckdb/node-api` read method names (Task 4) are the two API-surface spots to confirm against the installed versions at implementation time — adapt the unwrap/read calls, not the logic/SQL.
- **Ahrefs:** enum exists but no column mapping yet — out of scope (only semrush/mangools in v1; the `source` enum still accepts ahrefs for a future normalizer).
- **Object storage infra (MinIO on Dokploy)** is an ops task outside this repo plan; tests use `STORAGE_DRIVER=local`. Wire real S3 creds via env in Task 9/deploy.
- **Trend** parsing deferred (v1 sets `trend: []`); the rust path still handles trend for the small import.