feat: add watch history import from Trakt, Simkl, and Letterboxd (#13)

apps/public-api/.env.example

@@ -0,0 +1,6 @@
+TRAKT_CLIENT_ID=
+TRAKT_CLIENT_SECRET=
+SIMKL_CLIENT_ID=
+SIMKL_CLIENT_SECRET=
+GITHUB_TOKEN=
+POSTHOG_API_KEY=

apps/public-api/package.json

@@ -11,7 +11,12 @@
     "check-types": "tsc --noEmit"
   },
   "dependencies": {
-    "hono": "4.12.8"
+    "@hono/zod-validator": "0.7.6",
+    "@sofa/api": "workspace:*",
+    "@sofa/core": "workspace:*",
+    "@vercel/firewall": "1.1.2",
+    "hono": "4.12.8",
+    "zod": "catalog:"
   },
   "devDependencies": {
     "@types/bun": "catalog:",

apps/public-api/src/app.ts

@@ -1,5 +1,9 @@
+import { zValidator } from "@hono/zod-validator";
+import { checkRateLimit } from "@vercel/firewall";
 import { Hono } from "hono";
 import { cors } from "hono/cors";
+import { z } from "zod";
+import { getProvider, getProviderConfig } from "./providers";
 
 const GITHUB_RELEASES_URL =
   "https://api.github.com/repos/jakejarvis/sofa/releases/latest";
@@ -8,6 +12,8 @@ const app = new Hono();
 
 app.use("*", cors());
 
+// ─── Version Check ──────────────────────────────────────────
+
 app.get("/v1/version", async (c) => {
   try {
     const res = await fetch(GITHUB_RELEASES_URL, {
@@ -44,42 +50,179 @@ app.get("/v1/version", async (c) => {
   }
 });
 
-app.post("/v1/telemetry", async (c) => {
-  const body = await c.req.json();
+// ─── Telemetry ──────────────────────────────────────────────
 
-  if (!body.instanceId || !body.version) {
-    return c.json({ error: "Missing required fields" }, 400);
-  }
+app.post(
+  "/v1/telemetry",
+  zValidator(
+    "json",
+    z.object({
+      instanceId: z.string().min(1),
+      version: z.string().min(1),
+      arch: z.string().optional(),
+      users: z.number().optional(),
+      titles: z.number().optional(),
+      features: z.record(z.string(), z.unknown()).optional(),
+    }),
+  ),
+  async (c) => {
+    const body = c.req.valid("json");
+
+    const posthogKey = process.env.POSTHOG_API_KEY;
+    if (!posthogKey) {
+      return c.body(null, 204);
+    }
+
+    try {
+      await fetch("https://us.i.posthog.com/i/v0/e/", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          api_key: posthogKey,
+          event: "instance_report",
+          distinct_id: body.instanceId,
+          properties: {
+            version: body.version,
+            arch: body.arch,
+            users: body.users,
+            titles: body.titles,
+            ...(body.features ?? {}),
+          },
+        }),
+        signal: AbortSignal.timeout(10_000),
+      });
+    } catch {
+      // Fire-and-forget — don't fail the request if PostHog is down
+    }
 
-  const posthogKey = process.env.POSTHOG_API_KEY;
-  if (!posthogKey) {
-    // return okay, this isn't the user's problem
     return c.body(null, 204);
-  }
+  },
+);
 
-  try {
-    await fetch("https://us.i.posthog.com/i/v0/e/", {
-      method: "POST",
-      headers: { "Content-Type": "application/json" },
-      body: JSON.stringify({
-        api_key: posthogKey,
-        event: "instance_report",
-        distinct_id: body.instanceId,
-        properties: {
-          version: body.version,
-          arch: body.arch,
-          users: body.users,
-          titles: body.titles,
-          ...(body.features ?? {}),
+// ─── Import OAuth Proxy ─────────────────────────────────────
+// Rate limiting: Configure via Vercel dashboard WAF/rate-limit rules:
+//   - "import-device-code" — device-code initiation (e.g. 5 req/IP/300 sec)
+//   - "import-poll"        — token polling (e.g. 60 req/IP/minute)
+// The self-hosted server also prevents concurrent imports per user
+// via the importJobs table.
+
+const ProviderEnum = z.enum(["trakt", "simkl"]);
+
+app.post(
+  "/v1/import/:provider/device-code",
+  zValidator(
+    "param",
+    z.object({
+      provider: ProviderEnum,
+    }),
+  ),
+  async (c) => {
+    try {
+      const { rateLimited } = await checkRateLimit("import-device-code", {
+        request: c.req.raw,
+      });
+      if (rateLimited) {
+        return c.json({ error: "Rate limit exceeded" }, 429);
+      }
+    } catch (e) {
+      // Rate limiter error (e.g. WAF rule not configured) — fail open but log
+      console.warn("checkRateLimit error (import-device-code):", e);
+    }
+
+    const { provider: providerName } = c.req.valid("param");
+
+    const provider = getProvider(providerName);
+    const config = getProviderConfig(providerName);
+    if (!provider || !config.clientId) {
+      return c.json({ error: `${providerName} is not configured` }, 503);
+    }
+
+    try {
+      const result = await provider.getDeviceCode(
+        config.clientId,
+        config.clientSecret,
+      );
+      return c.json(result);
+    } catch (e) {
+      return c.json(
+        {
+          error: e instanceof Error ? e.message : "Failed to get device code",
         },
-      }),
-      signal: AbortSignal.timeout(10_000),
-    });
-  } catch {
-    // Fire-and-forget — don't fail the request if PostHog is down
-  }
+        502,
+      );
+    }
+  },
+);
 
-  return c.body(null, 204);
-});
+app.post(
+  "/v1/import/:provider/poll",
+  zValidator(
+    "param",
+    z.object({
+      provider: ProviderEnum,
+    }),
+  ),
+  zValidator(
+    "json",
+    z.object({
+      device_code: z.string().min(1).max(256),
+    }),
+  ),
+  async (c) => {
+    try {
+      const { rateLimited } = await checkRateLimit("import-poll", {
+        request: c.req.raw,
+      });
+      if (rateLimited) {
+        return c.json({ error: "Rate limit exceeded" }, 429);
+      }
+    } catch (e) {
+      // Rate limiter error (e.g. WAF rule not configured) — fail open but log
+      console.warn("checkRateLimit error (import-poll):", e);
+    }
+
+    const { provider: providerName } = c.req.valid("param");
+    const { device_code } = c.req.valid("json");
+
+    const provider = getProvider(providerName);
+    const config = getProviderConfig(providerName);
+    if (!provider || !config.clientId) {
+      return c.json({ error: `${providerName} is not configured` }, 503);
+    }
+
+    try {
+      const result = await provider.pollForToken(
+        config.clientId,
+        config.clientSecret,
+        device_code,
+      );
+
+      if (result.status !== "authorized") {
+        return c.json({ status: result.status });
+      }
+
+      // Fetch user data and return it inline
+      try {
+        const data = await provider.fetchUserData(
+          result.accessToken,
+          config.clientId,
+        );
+        return c.json({ status: "authorized", data });
+      } catch (e) {
+        // Auth succeeded but data fetch failed. Return a distinct status so
+        // the client can show a meaningful error instead of polling forever.
+        return c.json({
+          status: "fetch_error",
+          error: e instanceof Error ? e.message : "Failed to fetch user data",
+        });
+      }
+    } catch (e) {
+      return c.json(
+        { error: e instanceof Error ? e.message : "Poll failed" },
+        502,
+      );
+    }
+  },
+);
 
 export default app;

apps/public-api/src/providers/index.ts

@@ -0,0 +1,34 @@
+import { simkl } from "./simkl";
+import { trakt } from "./trakt";
+import type { ImportProvider } from "./types";
+
+const providers: Record<string, ImportProvider> = {
+  trakt,
+  simkl,
+};
+
+export function getProvider(name: string): ImportProvider | undefined {
+  return providers[name];
+}
+
+export function getProviderConfig(name: string): {
+  clientId: string;
+  clientSecret: string;
+} {
+  switch (name) {
+    case "trakt":
+      return {
+        clientId: process.env.TRAKT_CLIENT_ID ?? "",
+        clientSecret: process.env.TRAKT_CLIENT_SECRET ?? "",
+      };
+    case "simkl":
+      return {
+        clientId: process.env.SIMKL_CLIENT_ID ?? "",
+        clientSecret: process.env.SIMKL_CLIENT_SECRET ?? "",
+      };
+    default:
+      return { clientId: "", clientSecret: "" };
+  }
+}
+
+export type { DeviceCodeResponse, ImportProvider } from "./types";

apps/public-api/src/providers/simkl.ts

@@ -0,0 +1,163 @@
+import { parseSimklPayload } from "@sofa/core/imports/parsers";
+import type {
+  DeviceCodeResponse,
+  ImportProvider,
+  NormalizedImport,
+  PollResult,
+} from "./types";
+
+const API_BASE = "https://api.simkl.com";
+const AUTH_BASE = "https://simkl.com";
+
+function simklHeaders(
+  clientId: string,
+  token?: string,
+): Record<string, string> {
+  const headers: Record<string, string> = {
+    "Content-Type": "application/json",
+    "simkl-api-key": clientId,
+  };
+  if (token) headers.Authorization = `Bearer ${token}`;
+  return headers;
+}
+
+// ─── Simkl API types ─────────────────────────────────────────
+
+interface SimklIds {
+  imdb?: string;
+  tmdb?: string | number;
+  tvdb?: string | number;
+}
+
+interface SimklApiItem {
+  status?: string;
+  user_rating?: number;
+  last_watched_at?: string;
+  movie?: { title?: string; year?: number; ids?: SimklIds };
+  show?: { title?: string; year?: number; ids?: SimklIds };
+  seasons?: {
+    number?: number;
+    episodes?: { number?: number; watched_at?: string }[];
+  }[];
+}
+
+/** Flatten Simkl API response items into the shape parseSimklPayload expects.
+ *  When fetched with `episode_watched_at=yes`, the API returns ALL episodes
+ *  in the seasons array — filter to only those with a `watched_at` timestamp
+ *  so unwatched episodes don't get imported as watched. */
+function flattenSimklItems(items: SimklApiItem[], mediaKey: "movie" | "show") {
+  return items.map((item) => {
+    const media = item[mediaKey];
+    // Strip unwatched episodes from API response (they lack watched_at),
+    // then drop seasons that end up empty so the parser's missing-episode
+    // warning still fires correctly.
+    const filteredSeasons = item.seasons
+      ?.map((s) => ({
+        ...s,
+        episodes: s.episodes?.filter((ep) => ep.watched_at),
+      }))
+      .filter((s) => s.episodes && s.episodes.length > 0);
+    return {
+      title: media?.title,
+      year: media?.year,
+      ids: media?.ids,
+      status: item.status,
+      user_rating: item.user_rating,
+      last_watched_at: item.last_watched_at,
+      seasons: filteredSeasons,
+    };
+  });
+}
+
+// ─── Provider ────────────────────────────────────────────────
+
+export const simkl: ImportProvider = {
+  async getDeviceCode(clientId): Promise<DeviceCodeResponse> {
+    const res = await fetch(`${API_BASE}/oauth/pin?client_id=${clientId}`, {
+      method: "GET",
+      signal: AbortSignal.timeout(10_000),
+    });
+    if (!res.ok) throw new Error(`Simkl device code failed: ${res.status}`);
+    const data = (await res.json()) as {
+      device_code: string;
+      user_code: string;
+      verification_url: string;
+      expires_in: number;
+      interval: number;
+    };
+    return {
+      device_code: data.user_code, // Simkl checks PIN status by user_code, not device_code
+      user_code: data.user_code,
+      verification_url: data.verification_url || `${AUTH_BASE}/pin`,
+      expires_in: data.expires_in,
+      interval: data.interval || 5,
+    };
+  },
+
+  async pollForToken(clientId, _clientSecret, deviceCode): Promise<PollResult> {
+    const res = await fetch(`${API_BASE}/oauth/pin/${deviceCode}`, {
+      method: "GET",
+      headers: { "simkl-api-key": clientId },
+      signal: AbortSignal.timeout(10_000),
+    });
+
+    if (!res.ok) return { status: "pending" };
+
+    const data = (await res.json()) as {
+      result?: string;
+      access_token?: string;
+    };
+
+    if (data.result === "OK" && data.access_token) {
+      return { status: "authorized", accessToken: data.access_token };
+    }
+    if (data.result === "KO") return { status: "denied" };
+    return { status: "pending" };
+  },
+
+  async fetchUserData(accessToken, clientId): Promise<NormalizedImport> {
+    const headers = simklHeaders(clientId, accessToken);
+
+    // Fetch movies, shows, and anime in parallel
+    const [moviesRes, showsRes, animeRes] = await Promise.all([
+      fetch(`${API_BASE}/sync/all-items/movies`, { headers }),
+      fetch(
+        `${API_BASE}/sync/all-items/shows?extended=full&episode_watched_at=yes`,
+        { headers },
+      ),
+      fetch(
+        `${API_BASE}/sync/all-items/anime?extended=full&episode_watched_at=yes`,
+        { headers },
+      ),
+    ]);
+
+    // If all endpoints failed, throw so the caller gets a clear error
+    if (!moviesRes.ok && !showsRes.ok && !animeRes.ok) {
+      throw new Error(
+        `Simkl API returned errors: movies ${moviesRes.status}, shows ${showsRes.status}, anime ${animeRes.status}`,
+      );
+    }
+
+    const [moviesData, showsData, animeData] = await Promise.all([
+      moviesRes.ok
+        ? (moviesRes.json() as Promise<SimklApiItem[]>)
+        : ([] as SimklApiItem[]),
+      showsRes.ok
+        ? (showsRes.json() as Promise<SimklApiItem[]>)
+        : ([] as SimklApiItem[]),
+      animeRes.ok
+        ? (animeRes.json() as Promise<SimklApiItem[]>)
+        : ([] as SimklApiItem[]),
+    ]);
+
+    // Flatten API's nested movie/show objects into the flat format
+    // the core parser expects, then delegate normalization
+    const result = parseSimklPayload({
+      movies: flattenSimklItems(moviesData, "movie"),
+      shows: flattenSimklItems(showsData, "show"),
+      anime: flattenSimklItems(animeData, "show"),
+    });
+
+    return result.data;
+  },
+};

apps/public-api/src/providers/trakt.ts

@@ -0,0 +1,104 @@
+import { parseTraktPayload } from "@sofa/core/imports/parsers";
+import type {
+  DeviceCodeResponse,
+  ImportProvider,
+  NormalizedImport,
+  PollResult,
+} from "./types";
+
+const API_BASE = "https://api.trakt.tv";
+
+function traktHeaders(
+  clientId: string,
+  token?: string,
+): Record<string, string> {
+  const headers: Record<string, string> = {
+    "Content-Type": "application/json",
+    "trakt-api-version": "2",
+    "trakt-api-key": clientId,
+  };
+  if (token) headers.Authorization = `Bearer ${token}`;
+  return headers;
+}
+
+// ─── Provider ────────────────────────────────────────────────
+
+export const trakt: ImportProvider = {
+  async getDeviceCode(clientId): Promise<DeviceCodeResponse> {
+    const res = await fetch(`${API_BASE}/oauth/device/code`, {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ client_id: clientId }),
+      signal: AbortSignal.timeout(10_000),
+    });
+    if (!res.ok) throw new Error(`Trakt device code failed: ${res.status}`);
+    return (await res.json()) as DeviceCodeResponse;
+  },
+
+  async pollForToken(clientId, clientSecret, deviceCode): Promise<PollResult> {
+    const res = await fetch(`${API_BASE}/oauth/device/token`, {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        code: deviceCode,
+        client_id: clientId,
+        client_secret: clientSecret,
+      }),
+      signal: AbortSignal.timeout(10_000),
+    });
+
+    if (res.status === 200) {
+      const data = (await res.json()) as { access_token: string };
+      return { status: "authorized", accessToken: data.access_token };
+    }
+    if (res.status === 400) return { status: "pending" };
+    if (res.status === 404) return { status: "expired" };
+    if (res.status === 410) return { status: "expired" };
+    if (res.status === 418) return { status: "denied" };
+    if (res.status === 429) return { status: "pending" };
+    // 5xx: likely transient — keep polling
+    if (res.status >= 500) return { status: "pending" };
+    // Unknown 4xx: likely permanent — treat as expired
+    return { status: "expired" };
+  },
+
+  async fetchUserData(accessToken, clientId): Promise<NormalizedImport> {
+    const headers = traktHeaders(clientId, accessToken);
+
+    // Fetch all data in parallel
+    const [moviesRes, showsRes, watchlistRes, ratingsRes] = await Promise.all([
+      fetch(`${API_BASE}/sync/history/movies?limit=10000`, { headers }),
+      fetch(`${API_BASE}/sync/history/shows?limit=10000`, { headers }),
+      fetch(`${API_BASE}/sync/watchlist?extended=metadata&limit=10000`, {
+        headers,
+      }),
+      fetch(`${API_BASE}/sync/ratings`, { headers }),
+    ]);
+
+    // If all endpoints failed, throw so the caller gets a clear error
+    if (!moviesRes.ok && !showsRes.ok && !watchlistRes.ok && !ratingsRes.ok) {
+      throw new Error(
+        `Trakt API returned errors: movies ${moviesRes.status}, shows ${showsRes.status}, watchlist ${watchlistRes.status}, ratings ${ratingsRes.status}`,
+      );
+    }
+
+    const [moviesData, showsData, watchlistData, ratingsData] =
+      await Promise.all([
+        moviesRes.ok ? moviesRes.json() : [],
+        showsRes.ok ? showsRes.json() : [],
+        watchlistRes.ok ? watchlistRes.json() : [],
+        ratingsRes.ok ? ratingsRes.json() : [],
+      ]);
+
+    // Restructure API response into the format parseTraktPayload expects.
+    // The Trakt API returns the same item shapes as the JSON export format.
+    type TraktPayload = Parameters<typeof parseTraktPayload>[0];
+    const result = parseTraktPayload({
+      history: { movies: moviesData, shows: showsData },
+      watchlist: watchlistData,
+      ratings: ratingsData,
+    } as TraktPayload);
+
+    return result.data;
+  },
+};

apps/public-api/src/providers/types.ts

@@ -0,0 +1,33 @@
+import type { NormalizedImport } from "@sofa/api/schemas";
+
+export type { NormalizedImport };
+
+export interface DeviceCodeResponse {
+  device_code: string;
+  user_code: string;
+  verification_url: string;
+  expires_in: number;
+  interval: number;
+}
+
+export type PollResult =
+  | { status: "pending" }
+  | { status: "authorized"; accessToken: string }
+  | { status: "expired" }
+  | { status: "denied" };
+
+export interface ImportProvider {
+  getDeviceCode(
+    clientId: string,
+    clientSecret: string,
+  ): Promise<DeviceCodeResponse>;
+  pollForToken(
+    clientId: string,
+    clientSecret: string,
+    deviceCode: string,
+  ): Promise<PollResult>;
+  fetchUserData(
+    accessToken: string,
+    clientId: string,
+  ): Promise<NormalizedImport>;
+}