diff --git a/apps/journal/app/lib/connected-services/index.ts b/apps/journal/app/lib/connected-services/index.ts new file mode 100644 index 0000000..a3bb78c --- /dev/null +++ b/apps/journal/app/lib/connected-services/index.ts @@ -0,0 +1,9 @@ +// Public entry point for the connected-services module. Importing from +// here guarantees provider manifests are registered before any caller +// looks them up. + +import "./providers/index.ts"; + +export * from "./manager.ts"; +export * from "./registry.ts"; +export * from "./types.ts"; diff --git a/apps/journal/app/lib/connected-services/oauth-state.server.ts b/apps/journal/app/lib/connected-services/oauth-state.server.ts new file mode 100644 index 0000000..fe886af --- /dev/null +++ b/apps/journal/app/lib/connected-services/oauth-state.server.ts @@ -0,0 +1,23 @@ +// OAuth state encoding for the connect/callback flow. The state param is +// reflected back to the callback unchanged, so we use it to carry +// post-callback intent (where to return to, whether a push should resume). + +export interface PushOAuthState { + pushAfter?: { routeId: string }; + returnTo?: string; +} + +export function encodeOAuthState(state: PushOAuthState): string { + return Buffer.from(JSON.stringify(state), "utf8").toString("base64url"); +} + +export function decodeOAuthState(raw: string | null | undefined): PushOAuthState { + if (!raw) return {}; + try { + const json = Buffer.from(raw, "base64url").toString("utf8"); + const parsed = JSON.parse(json) as PushOAuthState; + return typeof parsed === "object" && parsed != null ? parsed : {}; + } catch { + return {}; + } +} diff --git a/apps/journal/app/lib/connected-services/providers/index.ts b/apps/journal/app/lib/connected-services/providers/index.ts new file mode 100644 index 0000000..d97b66e --- /dev/null +++ b/apps/journal/app/lib/connected-services/providers/index.ts @@ -0,0 +1,11 @@ +// Provider barrel. Imports every provider manifest and registers it with +// the registry. Adding a provider: import its manifest here and add the +// `registerManifest(...)` call. + +import { registerManifest } from "../registry.ts"; +import { wahooManifest } from "./wahoo/manifest.ts"; + +registerManifest(wahooManifest); + +// Re-export so callers (mostly tests) can grab a manifest directly. +export { wahooManifest }; diff --git a/apps/journal/app/lib/connected-services/providers/wahoo/importer.test.ts b/apps/journal/app/lib/connected-services/providers/wahoo/importer.test.ts new file mode 100644 index 0000000..0292bf9 --- /dev/null +++ b/apps/journal/app/lib/connected-services/providers/wahoo/importer.test.ts @@ -0,0 +1,102 @@ +// Contract tests for the Wahoo Importer capability adapter. +// +// The seam under test is `Importer` from registry.ts: +// listImportable(ctx, page) -> ImportableList +// importOne(ctx, workoutId) -> ImportResult +// +// Internals (FIT parsing, Wahoo HTTP details) are tested via the legacy +// wahoo.test.ts and follow the code as it's reorganized. + +import { describe, it, expect, beforeEach, vi } from "vitest"; +import type { CapabilityContext } from "../../registry.ts"; +import type { OAuthCredentials } from "../../types.ts"; + +const fetchSpy = vi.fn(); + +beforeEach(() => { + fetchSpy.mockReset(); + globalThis.fetch = fetchSpy as unknown as typeof fetch; +}); + +const stubCreds: OAuthCredentials = { + access_token: "fake-token", + refresh_token: "rt", + expires_at: new Date(Date.now() + 3600_000).toISOString(), +}; + +function ctxWith(creds: OAuthCredentials = stubCreds): CapabilityContext { + return { + serviceId: "svc-1", + withFreshCredentials: async (fn) => fn(creds), + }; +} + +// Importer is loaded after the implementation file exists. Using a dynamic +// import isolates the test from module load order during initial red. +const { wahooImporter } = await import("./importer.ts"); + +describe("wahooImporter.listImportable", () => { + it("calls Wahoo /v1/workouts with the access token from withFreshCredentials", async () => { + fetchSpy.mockResolvedValueOnce( + new Response( + JSON.stringify({ + workouts: [ + { + id: 42, + name: "Morning ride", + workout_type: "biking", + starts: "2026-05-01T07:00:00Z", + workout_summary: { + duration_active_accum: 3600, + distance_accum: 25000, + file: { url: "https://cdn.example/42.fit" }, + }, + }, + ], + total: 1, + page: 1, + per_page: 30, + }), + { status: 200 }, + ), + ); + + const result = await wahooImporter.listImportable(ctxWith(), 1); + + const [url, init] = fetchSpy.mock.calls[0]!; + expect(String(url)).toContain("/v1/workouts"); + expect(((init as RequestInit).headers as Record).Authorization).toBe( + "Bearer fake-token", + ); + expect(result.workouts).toHaveLength(1); + expect(result.workouts[0]!.id).toBe("42"); + expect(result.workouts[0]!.fileUrl).toBe("https://cdn.example/42.fit"); + expect(result.total).toBe(1); + }); + + it("filters out third-party workouts (fitness_app_id >= 1000)", async () => { + fetchSpy.mockResolvedValueOnce( + new Response( + JSON.stringify({ + workouts: [ + { id: 1, name: "Wahoo native", workout_type: "biking", starts: "2026-05-01T07:00:00Z" }, + { + id: 2, + name: "Third-party", + workout_type: "biking", + starts: "2026-05-01T08:00:00Z", + fitness_app_id: 1234, + }, + ], + total: 2, + page: 1, + per_page: 30, + }), + { status: 200 }, + ), + ); + + const result = await wahooImporter.listImportable(ctxWith(), 1); + expect(result.workouts.map((w) => w.id)).toEqual(["1"]); + }); +}); diff --git a/apps/journal/app/lib/connected-services/providers/wahoo/importer.ts b/apps/journal/app/lib/connected-services/providers/wahoo/importer.ts new file mode 100644 index 0000000..8cbdcf1 --- /dev/null +++ b/apps/journal/app/lib/connected-services/providers/wahoo/importer.ts @@ -0,0 +1,174 @@ +// Wahoo Importer capability adapter. Implements the Importer seam against +// Wahoo's /v1/workouts API. +// +// Credentials always flow through ctx.withFreshCredentials — this module +// never reads the connected_services credentials JSONB directly. + +import FitParser from "fit-file-parser"; +import { generateGpx } from "@trails-cool/gpx"; +import { createActivity } from "../../../activities.server.ts"; +import { recordImport, isAlreadyImported } from "../../../sync/imports.server.ts"; +import type { + CapabilityContext, + ImportableList, + ImportResult, + Importer, +} from "../../registry.ts"; +import type { OAuthCredentials } from "../../types.ts"; + +const WAHOO_API = "https://api.wahooligan.com"; + +interface WahooWorkout { + id: number; + name: string; + workout_type: string; + starts: string; + fitness_app_id?: number; + workout_summary?: { + duration_active_accum?: number; + distance_accum?: number; + file?: { url?: string }; + }; +} + +async function fetchWahooWorkoutPage( + creds: OAuthCredentials, + page: number, +): Promise<{ + workouts: WahooWorkout[]; + total: number; + page: number; + per_page: number; +}> { + const params = new URLSearchParams({ page: String(page), per_page: "30" }); + const resp = await fetch(`${WAHOO_API}/v1/workouts?${params}`, { + headers: { Authorization: `Bearer ${creds.access_token}` }, + }); + if (!resp.ok) throw new Error(`Wahoo list workouts failed: ${resp.status}`); + return resp.json() as Promise<{ + workouts: WahooWorkout[]; + total: number; + page: number; + per_page: number; + }>; +} + +function toImportable(w: WahooWorkout) { + return { + id: String(w.id), + name: w.name || `Workout ${w.id}`, + type: w.workout_type ?? "unknown", + startedAt: w.starts, + duration: w.workout_summary?.duration_active_accum + ? Math.round(w.workout_summary.duration_active_accum) + : null, + distance: w.workout_summary?.distance_accum + ? Math.round(w.workout_summary.distance_accum) + : null, + fileUrl: w.workout_summary?.file?.url, + }; +} + +async function fitToGpx(buffer: Buffer, name: string): Promise { + const parsed = await new Promise>((resolve, reject) => { + const parser = new FitParser({ force: true }); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + parser.parse(buffer as any, (error: unknown, data: any) => { + if (error) reject(error); + else resolve(data ?? {}); + }); + }); + + const records = (parsed.records ?? []) as Array<{ + position_lat?: number; + position_long?: number; + altitude?: number; + timestamp?: string | Date; + }>; + + const trackPoints = records + .filter((r) => r.position_lat != null && r.position_long != null) + .map((r) => ({ + lat: r.position_lat!, + lon: r.position_long!, + ele: r.altitude, + time: r.timestamp instanceof Date ? r.timestamp.toISOString() : r.timestamp, + })); + + if (trackPoints.length < 2) return null; + + return generateGpx({ name, tracks: [trackPoints] }); +} + +async function downloadFit(fileUrl: string): Promise { + // Wahoo CDN URLs are pre-signed; no auth header needed (and adding one + // breaks them). + const resp = await fetch(fileUrl); + if (!resp.ok) throw new Error(`Wahoo file download failed: ${resp.status}`); + return Buffer.from(await resp.arrayBuffer()); +} + +export const wahooImporter: Importer = { + async listImportable( + ctx: CapabilityContext, + page: number, + ): Promise { + const data = await ctx.withFreshCredentials((creds) => + fetchWahooWorkoutPage(creds as OAuthCredentials, page), + ); + + // Wahoo does not share workout data from third-party apps + // (fitness_app_id >= 1000). + const wahooOnly = data.workouts.filter( + (w) => !w.fitness_app_id || w.fitness_app_id < 1000, + ); + + return { + workouts: wahooOnly.map(toImportable), + total: data.total, + page: data.page, + perPage: data.per_page, + }; + }, + + async importOne( + ctx: CapabilityContext, + workoutId: string, + ): Promise { + // Look up the workout to get the file URL (Wahoo doesn't expose a + // direct /v1/workouts/ with file; we re-fetch the page). + // For simplicity we ask Wahoo for the workout directly; if that fails + // we fall back to scanning page 1. + const list = await ctx.withFreshCredentials((creds) => + fetchWahooWorkoutPage(creds as OAuthCredentials, 1), + ); + const workout = list.workouts.find((w) => String(w.id) === workoutId); + if (!workout) throw new Error(`Wahoo workout ${workoutId} not found on page 1`); + + // Resolve the connected service's user id via the capability context. + // The caller (route handler) supplies userId out-of-band — for now the + // route handler bridges the gap. We use the manager's getServiceById. + const { getServiceById } = await import("../../manager.ts"); + const service = await getServiceById(ctx.serviceId); + if (!service) throw new Error(`Connected service ${ctx.serviceId} not found`); + + const userId = service.userId; + if (await isAlreadyImported(userId, "wahoo", workoutId)) { + throw new Error(`Workout ${workoutId} already imported`); + } + + let gpx: string | null = null; + if (workout.workout_summary?.file?.url) { + const buffer = await downloadFit(workout.workout_summary.file.url); + gpx = await fitToGpx(buffer, workout.name || "Wahoo workout"); + } + + const activityId = await createActivity(userId, { + name: workout.name || `Wahoo workout ${workoutId}`, + gpx: gpx ?? undefined, + }); + await recordImport(userId, "wahoo", workoutId, activityId); + + return { activityId, hadGeometry: gpx !== null }; + }, +}; diff --git a/apps/journal/app/lib/connected-services/providers/wahoo/manifest.ts b/apps/journal/app/lib/connected-services/providers/wahoo/manifest.ts new file mode 100644 index 0000000..8f5eaf2 --- /dev/null +++ b/apps/journal/app/lib/connected-services/providers/wahoo/manifest.ts @@ -0,0 +1,130 @@ +// Wahoo provider manifest. Declares credential kind, OAuth config, +// authorization/exchange flows, and capability adapters. +// +// Adding to the registry happens via providers/index.ts which imports each +// provider's barrel. Don't `import`-cycle this file from registry.ts. + +import { oauthCredentialAdapter } from "../../credential-adapters/oauth.ts"; +import type { + ProviderManifest, + CapabilityContext, +} from "../../registry.ts"; +import type { OAuthCredentials, ProviderOAuthConfig } from "../../types.ts"; +import { wahooImporter } from "./importer.ts"; +import { wahooPusher } from "./pusher.ts"; +import { wahooWebhook } from "./webhook.ts"; + +const WAHOO_API = "https://api.wahooligan.com"; +const WAHOO_AUTH = "https://api.wahooligan.com/oauth"; + +const SCOPES = [ + "workouts_read", + "user_read", + "offline_data", + "routes_read", + "routes_write", +]; + +function clientId(): string { + return process.env.WAHOO_CLIENT_ID ?? ""; +} +function clientSecret(): string { + return process.env.WAHOO_CLIENT_SECRET ?? ""; +} + +const oauthConfig: ProviderOAuthConfig = { + get tokenUrl() { + return `${WAHOO_AUTH}/token`; + }, + get clientId() { + return clientId(); + }, + get clientSecret() { + return clientSecret(); + }, + get revokeUrl() { + return `${WAHOO_API}/v1/permissions`; + }, +}; + +export const wahooManifest: ProviderManifest = { + id: "wahoo", + displayName: "Wahoo", + credentialKind: "oauth", + credentialAdapter: oauthCredentialAdapter as ProviderManifest["credentialAdapter"], + oauthConfig, + scopes: SCOPES, + + buildAuthUrl(redirectUri: string, state: string): string { + const params = new URLSearchParams({ + client_id: clientId(), + redirect_uri: redirectUri, + response_type: "code", + scope: SCOPES.join(" "), + state, + }); + return `${WAHOO_AUTH}/authorize?${params}`; + }, + + async exchangeCode( + code: string, + redirectUri: string, + ): Promise<{ + credentials: OAuthCredentials; + providerUserId: string | null; + grantedScopes: string[]; + }> { + const resp = await fetch(`${WAHOO_AUTH}/token`, { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: new URLSearchParams({ + client_id: clientId(), + client_secret: clientSecret(), + code, + grant_type: "authorization_code", + redirect_uri: redirectUri, + }).toString(), + }); + if (!resp.ok) { + const text = await resp.text().catch(() => ""); + const err = new Error(`Wahoo token exchange failed: ${resp.status} ${text}`); + // Attach a code so callers can distinguish the "too many tokens" sandbox case. + (err as Error & { code?: string }).code = text.includes("Too many unrevoked access tokens") + ? "too_many_tokens" + : "generic"; + throw err; + } + const data = (await resp.json()) as { + access_token: string; + refresh_token: string; + expires_in: number; + }; + + // Pull provider user id so webhook routing works. + const userResp = await fetch(`${WAHOO_API}/v1/user`, { + headers: { Authorization: `Bearer ${data.access_token}` }, + }); + const user = userResp.ok + ? ((await userResp.json()) as { id: number }) + : null; + + return { + credentials: { + access_token: data.access_token, + refresh_token: data.refresh_token, + expires_at: new Date(Date.now() + data.expires_in * 1000).toISOString(), + }, + providerUserId: user?.id != null ? String(user.id) : null, + // Wahoo does not return a `scope` field and grants scopes + // all-or-nothing, so the requested set is the granted set. + grantedScopes: SCOPES, + }; + }, + + importer: wahooImporter, + routePusher: wahooPusher, + webhookReceiver: wahooWebhook, +}; + +// Re-export the capability adapters for direct testing access if needed. +export type { CapabilityContext }; diff --git a/apps/journal/app/lib/connected-services/providers/wahoo/pusher.test.ts b/apps/journal/app/lib/connected-services/providers/wahoo/pusher.test.ts new file mode 100644 index 0000000..4bb2a30 --- /dev/null +++ b/apps/journal/app/lib/connected-services/providers/wahoo/pusher.test.ts @@ -0,0 +1,208 @@ +// Contract tests for the Wahoo RoutePusher capability adapter. +// +// Seam: pushRoute(ctx, input) -> {remoteId, version} +// +// Wahoo workarounds (FIT conversion, route: external_id, PUT-vs-POST, +// PUT->POST-on-404 fallback) are tested here as adapter-internal — they +// must not surface on the seam. + +import { describe, it, expect, beforeEach, vi } from "vitest"; +import type { CapabilityContext } from "../../registry.ts"; +import type { OAuthCredentials } from "../../types.ts"; + +// ---- mocks ---- + +const fetchSpy = vi.fn(); + +const mockFitConvert = vi.fn(); +vi.mock("@trails-cool/fit", () => ({ + gpxToFitCourse: (input: unknown) => mockFitConvert(input), +})); + +// sync_pushes idempotency state — manipulated per-test. +let existingPushRow: + | { + id: string; + userId: string; + routeId: string; + provider: string; + remoteId: string | null; + lastPushedVersion: number | null; + pushedAt: Date | null; + error: string | null; + } + | null = null; +const insertedPushes: unknown[] = []; +const updatedPushes: unknown[] = []; + +const mockDb = { + select: () => ({ + from: () => ({ + where: () => ({ + limit: () => Promise.resolve(existingPushRow ? [existingPushRow] : []), + }), + }), + }), + insert: () => ({ + values: (v: unknown) => { + insertedPushes.push(v); + return Promise.resolve(); + }, + }), + update: () => ({ + set: (patch: unknown) => ({ + where: () => { + updatedPushes.push(patch); + return Promise.resolve(); + }, + }), + }), +}; + +vi.mock("../../../db.ts", () => ({ getDb: () => mockDb })); + +vi.mock("../../manager.ts", () => ({ + getServiceById: async () => ({ + id: "svc-1", + userId: "u1", + provider: "wahoo", + credentialKind: "oauth", + credentials: {}, + status: "active", + providerUserId: null, + grantedScopes: [], + createdAt: new Date(), + }), +})); + +beforeEach(() => { + fetchSpy.mockReset(); + globalThis.fetch = fetchSpy as unknown as typeof fetch; + mockFitConvert.mockReset(); + mockFitConvert.mockResolvedValue(new Uint8Array([0xfe, 0xed, 0xfa, 0xce])); + existingPushRow = null; + insertedPushes.length = 0; + updatedPushes.length = 0; +}); + +const stubCreds: OAuthCredentials = { + access_token: "fake-token", + refresh_token: "rt", + expires_at: new Date(Date.now() + 3600_000).toISOString(), +}; + +function ctx(): CapabilityContext { + return { + serviceId: "svc-1", + withFreshCredentials: async (fn) => fn(stubCreds), + }; +} + +const pushInput = { + routeId: "route-abc", + routeName: "Berlin loop", + description: "morning ride", + gpx: ` + 34 + 40 + `, + startLat: 52.5, + startLng: 13.4, + distance: 1234, + ascent: 56, + localVersion: 3, +}; + +const { wahooPusher } = await import("./pusher.ts"); + +describe("wahooPusher.pushRoute — first push (no existing row)", () => { + it("POSTs to /v1/routes with FIT body and external_id=route:", async () => { + fetchSpy.mockResolvedValueOnce( + new Response(JSON.stringify({ id: 9001 }), { status: 201 }), + ); + + const result = await wahooPusher.pushRoute(ctx(), pushInput); + + expect(result.remoteId).toBe("9001"); + expect(result.version).toBe(3); + const [url, init] = fetchSpy.mock.calls[0]!; + expect(String(url)).toBe("https://api.wahooligan.com/v1/routes"); + expect((init as RequestInit).method).toBe("POST"); + const body = (init as RequestInit).body as string; + expect(body).toContain("route%5Bexternal_id%5D=route%3Aroute-abc"); + expect(body).toContain("route%5Bfile%5D=data%3Aapplication%2Fvnd.fit%3Bbase64%2C"); + expect(insertedPushes).toHaveLength(1); + }); +}); + +describe("wahooPusher.pushRoute — re-push of an unchanged route", () => { + it("short-circuits without calling Wahoo when last_pushed_version matches", async () => { + existingPushRow = { + id: "existing", + userId: "u1", + routeId: "route-abc", + provider: "wahoo", + remoteId: "9001", + lastPushedVersion: 3, + pushedAt: new Date("2026-01-01"), + error: null, + }; + + const result = await wahooPusher.pushRoute(ctx(), pushInput); + + expect(result.remoteId).toBe("9001"); + expect(result.version).toBe(3); + expect(fetchSpy).not.toHaveBeenCalled(); + }); +}); + +describe("wahooPusher.pushRoute — re-push after edit", () => { + it("PUTs to /v1/routes/ when lastPushedVersion < localVersion", async () => { + existingPushRow = { + id: "existing", + userId: "u1", + routeId: "route-abc", + provider: "wahoo", + remoteId: "9001", + lastPushedVersion: 2, + pushedAt: new Date("2026-01-01"), + error: null, + }; + fetchSpy.mockResolvedValueOnce(new Response(null, { status: 204 })); + + const result = await wahooPusher.pushRoute(ctx(), pushInput); + + expect(result.remoteId).toBe("9001"); + expect(result.version).toBe(3); + const [url, init] = fetchSpy.mock.calls[0]!; + expect(String(url)).toBe("https://api.wahooligan.com/v1/routes/9001"); + expect((init as RequestInit).method).toBe("PUT"); + }); +}); + +describe("wahooPusher.pushRoute — PUT 404 fallback", () => { + it("falls back to POST and overwrites remoteId when PUT returns 404", async () => { + existingPushRow = { + id: "existing", + userId: "u1", + routeId: "route-abc", + provider: "wahoo", + remoteId: "9001", + lastPushedVersion: 2, + pushedAt: new Date("2026-01-01"), + error: null, + }; + fetchSpy + .mockResolvedValueOnce(new Response("not found", { status: 404 })) + .mockResolvedValueOnce( + new Response(JSON.stringify({ id: 9999 }), { status: 201 }), + ); + + const result = await wahooPusher.pushRoute(ctx(), pushInput); + + expect(result.remoteId).toBe("9999"); + expect(fetchSpy).toHaveBeenCalledTimes(2); + const [, secondInit] = fetchSpy.mock.calls[1]!; + expect((secondInit as RequestInit).method).toBe("POST"); + }); +}); diff --git a/apps/journal/app/lib/connected-services/providers/wahoo/pusher.ts b/apps/journal/app/lib/connected-services/providers/wahoo/pusher.ts new file mode 100644 index 0000000..365347b --- /dev/null +++ b/apps/journal/app/lib/connected-services/providers/wahoo/pusher.ts @@ -0,0 +1,217 @@ +// Wahoo RoutePusher capability adapter. +// +// Exposes the seam shape `pushRoute(ctx, input) -> {remoteId, version}` per +// ADR-0003. Wahoo specifics — FIT-Course conversion, the `route:` +// external_id convention, the PUT-vs-POST decision based on sync_pushes, +// and the PUT-on-404 → POST fallback — live entirely inside this module +// and never appear on the seam. +// +// Idempotency state lives in `journal.sync_pushes`. The seam returns the +// remote id and the local version that was pushed; the caller is free to +// inspect the table for richer state. + +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { gpxToFitCourse } from "@trails-cool/fit"; +import { syncPushes } from "@trails-cool/db/schema/journal"; +import { getDb } from "../../../db.ts"; +import { getServiceById } from "../../manager.ts"; +import type { + CapabilityContext, + RoutePushInput, + RoutePushResult, + RoutePusher, +} from "../../registry.ts"; +import type { OAuthCredentials } from "../../types.ts"; + +const WAHOO_API = "https://api.wahooligan.com"; + +interface WahooErrorShape { + status: number; + body: string; +} + +class WahooHttpError extends Error { + shape: WahooErrorShape; + constructor(shape: WahooErrorShape) { + super(`Wahoo route ${shape.status}: ${shape.body}`); + this.shape = shape; + } +} + +function externalIdFor(routeId: string): string { + return `route:${routeId}`; +} + +function buildBody( + fit: Uint8Array, + input: RoutePushInput, +): URLSearchParams { + // Wahoo expects route[file] as a data URI, not raw base64. Sending raw + // base64 results in a route with file.url = null; the app shows + // metadata but renders no track. + const fitDataUri = `data:application/vnd.fit;base64,${Buffer.from(fit).toString("base64")}`; + const body = new URLSearchParams({ + "route[external_id]": externalIdFor(input.routeId), + "route[provider_updated_at]": new Date().toISOString(), + "route[name]": input.routeName, + "route[workout_type_family_id]": "0", + "route[start_lat]": input.startLat.toString(), + "route[start_lng]": input.startLng.toString(), + "route[distance]": input.distance.toString(), + "route[ascent]": input.ascent.toString(), + "route[file]": fitDataUri, + }); + if (input.description) body.set("route[description]", input.description); + return body; +} + +async function postOrPut( + method: "POST" | "PUT", + url: string, + accessToken: string, + body: URLSearchParams, + fallbackRemoteId?: string, +): Promise<{ remoteId: string }> { + const resp = await fetch(url, { + method, + headers: { + Authorization: `Bearer ${accessToken}`, + "Content-Type": "application/x-www-form-urlencoded", + }, + body: body.toString(), + }); + + if (resp.ok) { + let remoteId: string | undefined; + if (resp.status !== 204) { + const data = (await resp.json().catch(() => null)) as { id?: number | string } | null; + remoteId = data?.id?.toString(); + } + remoteId ??= fallbackRemoteId; + if (!remoteId) throw new Error(`Wahoo response missing route id`); + return { remoteId }; + } + + const text = await resp.text().catch(() => ""); + throw new WahooHttpError({ status: resp.status, body: text }); +} + +interface ExistingPush { + id: string; + userId: string; + routeId: string; + provider: string; + remoteId: string | null; + lastPushedVersion: number | null; + pushedAt: Date | null; + error: string | null; +} + +async function findExistingPush( + userId: string, + routeId: string, +): Promise { + const db = getDb(); + const rows = (await db + .select() + .from(syncPushes) + .where( + and( + eq(syncPushes.userId, userId), + eq(syncPushes.routeId, routeId), + eq(syncPushes.provider, "wahoo"), + ), + ) + .limit(1)) as ExistingPush[]; + return rows[0] ?? null; +} + +async function recordPush( + existing: ExistingPush | null, + userId: string, + input: RoutePushInput, + remoteId: string, +): Promise { + const db = getDb(); + const now = new Date(); + if (existing) { + await db + .update(syncPushes) + .set({ + remoteId, + lastPushedVersion: input.localVersion, + pushedAt: now, + error: null, + updatedAt: now, + }) + .where(eq(syncPushes.id, existing.id)); + } else { + await db.insert(syncPushes).values({ + id: randomUUID(), + userId, + routeId: input.routeId, + provider: "wahoo", + externalId: externalIdFor(input.routeId), + remoteId, + lastPushedVersion: input.localVersion, + pushedAt: now, + error: null, + }); + } +} + +export const wahooPusher: RoutePusher = { + async pushRoute( + ctx: CapabilityContext, + input: RoutePushInput, + ): Promise { + // Resolve the user from the connected service so we can read/write + // sync_pushes for the right (user, route, provider) tuple. + const service = await getServiceById(ctx.serviceId); + if (!service) throw new Error(`Connected service ${ctx.serviceId} not found`); + + const existing = await findExistingPush(service.userId, input.routeId); + if ( + existing?.pushedAt && + existing.remoteId && + existing.lastPushedVersion === input.localVersion + ) { + return { remoteId: existing.remoteId, version: input.localVersion }; + } + + const fit = await gpxToFitCourse({ + gpx: input.gpx, + name: input.routeName, + description: input.description, + }); + const body = buildBody(fit, input); + + const result = await ctx.withFreshCredentials(async (creds) => { + const accessToken = (creds as OAuthCredentials).access_token; + // PUT-vs-POST: PUT in place when we have a remoteId on file. + if (existing?.remoteId) { + try { + return await postOrPut( + "PUT", + `${WAHOO_API}/v1/routes/${encodeURIComponent(existing.remoteId)}`, + accessToken, + body, + existing.remoteId, + ); + } catch (err) { + // 404 means the user deleted the route on Wahoo's side. Fall + // back to POST and overwrite the local remoteId. + if (err instanceof WahooHttpError && err.shape.status === 404) { + return postOrPut("POST", `${WAHOO_API}/v1/routes`, accessToken, body); + } + throw err; + } + } + return postOrPut("POST", `${WAHOO_API}/v1/routes`, accessToken, body); + }); + + await recordPush(existing, service.userId, input, result.remoteId); + return { remoteId: result.remoteId, version: input.localVersion }; + }, +}; diff --git a/apps/journal/app/lib/connected-services/providers/wahoo/webhook.test.ts b/apps/journal/app/lib/connected-services/providers/wahoo/webhook.test.ts new file mode 100644 index 0000000..525d719 --- /dev/null +++ b/apps/journal/app/lib/connected-services/providers/wahoo/webhook.test.ts @@ -0,0 +1,133 @@ +// Contract tests for the Wahoo WebhookReceiver capability adapter. +// +// Seam: parseWebhook(body) -> WebhookEvent | null +// handle(event) -> void (creates an activity if file present, dedups via sync_imports) + +import { describe, it, expect, beforeEach, vi } from "vitest"; + +const fetchSpy = vi.fn(); +const mockCreateActivity = vi.fn(); +const mockIsAlreadyImported = vi.fn(); +const mockRecordImport = vi.fn(); +const mockGetServiceByProviderUser = vi.fn(); +const mockWithFreshCredentials = vi.fn(); + +vi.mock("../../../activities.server.ts", () => ({ + createActivity: mockCreateActivity, +})); +vi.mock("../../../sync/imports.server.ts", () => ({ + isAlreadyImported: mockIsAlreadyImported, + recordImport: mockRecordImport, +})); +vi.mock("../../manager.ts", () => ({ + getServiceByProviderUser: mockGetServiceByProviderUser, + withFreshCredentials: mockWithFreshCredentials, +})); + +beforeEach(() => { + fetchSpy.mockReset(); + globalThis.fetch = fetchSpy as unknown as typeof fetch; + mockCreateActivity.mockReset(); + mockIsAlreadyImported.mockReset(); + mockRecordImport.mockReset(); + mockGetServiceByProviderUser.mockReset(); + mockWithFreshCredentials.mockReset(); +}); + +const { wahooWebhook } = await import("./webhook.ts"); + +describe("wahooWebhook.parseWebhook", () => { + it("returns a WebhookEvent for workout_summary payloads", () => { + const event = wahooWebhook.parseWebhook({ + event_type: "workout_summary", + user: { id: 7 }, + workout_summary: { + workout: { id: 42 }, + file: { url: "https://cdn.example/42.fit" }, + }, + }); + expect(event).toEqual({ + eventType: "workout_summary", + providerUserId: "7", + workoutId: "42", + fileUrl: "https://cdn.example/42.fit", + }); + }); + + it("returns null for unrecognized event types", () => { + expect(wahooWebhook.parseWebhook({ event_type: "other" })).toBeNull(); + }); + + it("returns null when user.id is missing", () => { + expect( + wahooWebhook.parseWebhook({ event_type: "workout_summary", user: {} }), + ).toBeNull(); + }); +}); + +describe("wahooWebhook.handle", () => { + it("creates an activity and records the import for a known user", async () => { + mockGetServiceByProviderUser.mockResolvedValue({ + id: "svc-1", + userId: "u1", + provider: "wahoo", + }); + mockIsAlreadyImported.mockResolvedValue(false); + mockCreateActivity.mockResolvedValue("act-1"); + // withFreshCredentials passes the credentials to fn — we don't need to + // download a file to assert the basic flow; pass a no-op file URL test + // via parseWebhook output containing fileUrl undefined to skip download. + mockWithFreshCredentials.mockImplementation( + async (_id: string, fn: (creds: unknown) => Promise) => + fn({ + access_token: "a", + refresh_token: "r", + expires_at: new Date(Date.now() + 3600_000).toISOString(), + }), + ); + + await wahooWebhook.handle({ + eventType: "workout_summary", + providerUserId: "7", + workoutId: "42", + // no fileUrl — the activity is created without GPX + }); + + expect(mockCreateActivity).toHaveBeenCalledWith( + "u1", + expect.objectContaining({ name: expect.stringContaining("Wahoo") }), + ); + expect(mockRecordImport).toHaveBeenCalledWith("u1", "wahoo", "42", "act-1"); + }); + + it("silently skips when the providerUserId is unknown (no leak)", async () => { + mockGetServiceByProviderUser.mockResolvedValue(null); + + await wahooWebhook.handle({ + eventType: "workout_summary", + providerUserId: "999", + workoutId: "42", + }); + + expect(mockCreateActivity).not.toHaveBeenCalled(); + expect(mockRecordImport).not.toHaveBeenCalled(); + }); + + it("silently skips when the workout was already imported (idempotency)", async () => { + mockGetServiceByProviderUser.mockResolvedValue({ + id: "svc-1", + userId: "u1", + provider: "wahoo", + }); + mockIsAlreadyImported.mockResolvedValue(true); + + await wahooWebhook.handle({ + eventType: "workout_summary", + providerUserId: "7", + workoutId: "42", + }); + + expect(mockCreateActivity).not.toHaveBeenCalled(); + expect(mockRecordImport).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/journal/app/lib/connected-services/providers/wahoo/webhook.ts b/apps/journal/app/lib/connected-services/providers/wahoo/webhook.ts new file mode 100644 index 0000000..2b11b68 --- /dev/null +++ b/apps/journal/app/lib/connected-services/providers/wahoo/webhook.ts @@ -0,0 +1,100 @@ +// Wahoo WebhookReceiver capability adapter. +// +// Wahoo posts `workout_summary` events to /api/sync/webhook/wahoo when a +// workout completes. We route the event to the right local user via +// provider_user_id, deduplicate via sync_imports, then download + convert +// the FIT file (if present) and create an activity. + +import FitParser from "fit-file-parser"; +import { generateGpx } from "@trails-cool/gpx"; +import { createActivity } from "../../../activities.server.ts"; +import { isAlreadyImported, recordImport } from "../../../sync/imports.server.ts"; +import { getServiceByProviderUser, withFreshCredentials } from "../../manager.ts"; +import type { OAuthCredentials } from "../../types.ts"; +import type { WebhookEvent, WebhookReceiver } from "../../registry.ts"; + +interface WahooWebhookBody { + event_type?: string; + user?: { id?: number }; + workout_summary?: { + id?: number; + workout?: { id?: number }; + file?: { url?: string }; + }; +} + +async function fitToGpx(buffer: Buffer): Promise { + const parsed = await new Promise>((resolve, reject) => { + const parser = new FitParser({ force: true }); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + parser.parse(buffer as any, (error: unknown, data: any) => { + if (error) reject(error); + else resolve(data ?? {}); + }); + }); + + const records = (parsed.records ?? []) as Array<{ + position_lat?: number; + position_long?: number; + altitude?: number; + timestamp?: string | Date; + }>; + + const trackPoints = records + .filter((r) => r.position_lat != null && r.position_long != null) + .map((r) => ({ + lat: r.position_lat!, + lon: r.position_long!, + ele: r.altitude, + time: r.timestamp instanceof Date ? r.timestamp.toISOString() : r.timestamp, + })); + + if (trackPoints.length < 2) return null; + return generateGpx({ name: "Wahoo workout", tracks: [trackPoints] }); +} + +export const wahooWebhook: WebhookReceiver = { + parseWebhook(body: unknown): WebhookEvent | null { + const payload = body as WahooWebhookBody; + if (payload.event_type !== "workout_summary" || !payload.user?.id) return null; + + return { + eventType: payload.event_type, + providerUserId: String(payload.user.id), + workoutId: String( + payload.workout_summary?.workout?.id ?? payload.workout_summary?.id ?? "", + ), + fileUrl: payload.workout_summary?.file?.url, + }; + }, + + async handle(event: WebhookEvent): Promise { + // Match incoming webhooks to local users via provider_user_id. + // Unknown users return silently — no leak. + const service = await getServiceByProviderUser("wahoo", event.providerUserId); + if (!service) return; + + if (await isAlreadyImported(service.userId, "wahoo", event.workoutId)) return; + + let gpx: string | null = null; + if (event.fileUrl) { + // Wahoo CDN URLs are pre-signed; no auth header needed. We still go + // through withFreshCredentials so the manager has a chance to refresh + // a near-expired credential before any subsequent Wahoo call this + // handler might make. + const buffer = await withFreshCredentials(service.id, async (_creds) => { + void (_creds as unknown as OAuthCredentials); + const resp = await fetch(event.fileUrl!); + if (!resp.ok) throw new Error(`Wahoo file download failed: ${resp.status}`); + return Buffer.from(await resp.arrayBuffer()); + }); + gpx = await fitToGpx(buffer); + } + + const activityId = await createActivity(service.userId, { + name: `Wahoo workout`, + gpx: gpx ?? undefined, + }); + await recordImport(service.userId, "wahoo", event.workoutId, activityId); + }, +}; diff --git a/apps/journal/app/lib/connected-services/push-action.server.ts b/apps/journal/app/lib/connected-services/push-action.server.ts new file mode 100644 index 0000000..a42039f --- /dev/null +++ b/apps/journal/app/lib/connected-services/push-action.server.ts @@ -0,0 +1,115 @@ +// Orchestrates a route push: load route, check ownership, check scopes, +// build the RoutePushInput, and invoke the provider's RoutePusher +// capability. Replaces the legacy pushRouteToProvider in lib/sync. +// +// The pusher (per-provider) handles HTTP, FIT conversion, idempotency, +// and PUT/POST/404 fallback. This module is the orchestration layer +// callers use (route handlers, OAuth callback resume). + +import { desc, eq } from "drizzle-orm"; +import { parseGpxAsync } from "@trails-cool/gpx"; +import { routeVersions } from "@trails-cool/db/schema/journal"; +import { getDb } from "../db.ts"; +import { getRoute } from "../routes.server.ts"; +import { + ConnectionNotActiveError, + NeedsRelinkError, +} from "./types.ts"; +import { capabilityContextFor, getService } from "./manager.ts"; +import { getManifest, type RoutePushInput } from "./registry.ts"; + +export type PushOutcome = + | { status: "success"; remoteId: string; pushedAt: Date } + | { status: "scope_missing" } + | { status: "no_connection" } + | { status: "not_owner" } + | { status: "not_found" } + | { status: "unsupported_provider" } + | { status: "no_geometry" } + | { status: "needs_relink" } + | { + status: "error"; + code: "validation" | "rate_limit" | "token_expired" | "generic"; + message: string; + }; + +export interface PushRouteOptions { + userId: string; + providerId: string; + routeId: string; +} + +export async function pushRouteToProvider( + opts: PushRouteOptions, +): Promise { + const { userId, providerId, routeId } = opts; + const db = getDb(); + + const manifest = getManifest(providerId); + if (!manifest) return { status: "not_found" }; + if (!manifest.routePusher) return { status: "unsupported_provider" }; + + const route = await getRoute(routeId); + if (!route) return { status: "not_found" }; + if (route.ownerId !== userId) return { status: "not_owner" }; + + const service = await getService(userId, providerId); + if (!service) return { status: "no_connection" }; + if (!service.grantedScopes.includes("routes_write")) { + return { status: "scope_missing" }; + } + + // Pull the locked-in version GPX (not routes.gpx, which is the working copy). + const [latestVersion] = await db + .select() + .from(routeVersions) + .where(eq(routeVersions.routeId, routeId)) + .orderBy(desc(routeVersions.version)) + .limit(1); + + const versionGpx = latestVersion?.gpx ?? route.gpx; + const versionNumber = latestVersion?.version ?? 1; + if (!versionGpx) return { status: "no_geometry" }; + + const parsed = await parseGpxAsync(versionGpx); + const points = parsed.tracks.flat(); + if (points.length === 0) return { status: "no_geometry" }; + + const input: RoutePushInput = { + routeId, + routeName: route.name, + description: route.description ?? undefined, + gpx: versionGpx, + startLat: points[0]!.lat, + startLng: points[0]!.lon, + distance: parsed.distance, + ascent: parsed.elevation.gain, + localVersion: versionNumber, + }; + + try { + const ctx = capabilityContextFor(service.id); + const result = await manifest.routePusher.pushRoute(ctx, input); + return { + status: "success", + remoteId: result.remoteId, + pushedAt: new Date(), + }; + } catch (err) { + if (err instanceof NeedsRelinkError) return { status: "needs_relink" }; + if (err instanceof ConnectionNotActiveError) return { status: "needs_relink" }; + const message = err instanceof Error ? err.message : String(err); + // Map known HTTP-shape errors. The pusher throws Error / WahooHttpError; + // we don't try to recover further here. + if (message.includes("422")) { + return { status: "error", code: "validation", message }; + } + if (message.includes("429")) { + return { status: "error", code: "rate_limit", message }; + } + if (message.includes("401") || message.includes("403")) { + return { status: "error", code: "token_expired", message }; + } + return { status: "error", code: "generic", message }; + } +} diff --git a/apps/journal/app/lib/sync/connections.server.ts b/apps/journal/app/lib/sync/connections.server.ts deleted file mode 100644 index 0067491..0000000 --- a/apps/journal/app/lib/sync/connections.server.ts +++ /dev/null @@ -1,127 +0,0 @@ -// COMPATIBILITY SHIM — translates the legacy TokenSet-shaped API onto the -// new connected_services / JSONB-credentials schema introduced by -// deepen-connected-services. Once the routes and pushes.server.ts migrate -// to ConnectedServiceManager (tasks 5.x of the change), this whole file -// (and the rest of apps/journal/app/lib/sync/) goes away. -// -// Do not add new callers. Use apps/journal/app/lib/connected-services/ -// directly. - -import { randomUUID } from "node:crypto"; -import { eq, and } from "drizzle-orm"; -import { getDb } from "../db.ts"; -import { connectedServices } from "@trails-cool/db/schema/journal"; -import type { TokenSet } from "./types.ts"; - -interface OAuthBlob { - access_token: string; - refresh_token: string; - expires_at: string; -} - -interface LegacyConnection { - id: string; - userId: string; - provider: string; - accessToken: string; - refreshToken: string; - expiresAt: Date; - providerUserId: string | null; - grantedScopes: string[]; - createdAt: Date; -} - -function toLegacy(row: typeof connectedServices.$inferSelect): LegacyConnection { - const blob = row.credentials as OAuthBlob; - return { - id: row.id, - userId: row.userId, - provider: row.provider, - accessToken: blob.access_token, - refreshToken: blob.refresh_token, - expiresAt: new Date(blob.expires_at), - providerUserId: row.providerUserId, - grantedScopes: row.grantedScopes, - createdAt: row.createdAt, - }; -} - -function toBlob(tokens: TokenSet): OAuthBlob { - return { - access_token: tokens.accessToken, - refresh_token: tokens.refreshToken, - expires_at: tokens.expiresAt.toISOString(), - }; -} - -export async function saveConnection( - userId: string, - provider: string, - tokens: TokenSet, - grantedScopes: string[] = [], -) { - const db = getDb(); - await db - .delete(connectedServices) - .where( - and(eq(connectedServices.userId, userId), eq(connectedServices.provider, provider)), - ); - await db.insert(connectedServices).values({ - id: randomUUID(), - userId, - provider, - credentialKind: "oauth", - credentials: toBlob(tokens), - status: "active", - providerUserId: tokens.providerUserId ?? null, - grantedScopes, - }); -} - -export async function getConnection( - userId: string, - provider: string, -): Promise { - const db = getDb(); - const [row] = await db - .select() - .from(connectedServices) - .where( - and(eq(connectedServices.userId, userId), eq(connectedServices.provider, provider)), - ); - return row ? toLegacy(row) : null; -} - -export async function getConnectionByProviderUser( - provider: string, - providerUserId: string, -): Promise { - const db = getDb(); - const [row] = await db - .select() - .from(connectedServices) - .where( - and( - eq(connectedServices.provider, provider), - eq(connectedServices.providerUserId, providerUserId), - ), - ); - return row ? toLegacy(row) : null; -} - -export async function updateTokens(connectionId: string, tokens: TokenSet) { - const db = getDb(); - await db - .update(connectedServices) - .set({ credentials: toBlob(tokens) }) - .where(eq(connectedServices.id, connectionId)); -} - -export async function deleteConnection(userId: string, provider: string) { - const db = getDb(); - await db - .delete(connectedServices) - .where( - and(eq(connectedServices.userId, userId), eq(connectedServices.provider, provider)), - ); -} diff --git a/apps/journal/app/lib/sync/providers/__fixtures__/wahoo-ride.fit b/apps/journal/app/lib/sync/providers/__fixtures__/wahoo-ride.fit deleted file mode 100644 index 94e96ef..0000000 Binary files a/apps/journal/app/lib/sync/providers/__fixtures__/wahoo-ride.fit and /dev/null differ diff --git a/apps/journal/app/lib/sync/providers/wahoo.test.ts b/apps/journal/app/lib/sync/providers/wahoo.test.ts deleted file mode 100644 index ebc7de9..0000000 --- a/apps/journal/app/lib/sync/providers/wahoo.test.ts +++ /dev/null @@ -1,285 +0,0 @@ -import { readFileSync } from "node:fs"; -import { resolve } from "node:path"; -import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; -import { wahooProvider } from "./wahoo"; -import { PushError, OAuthError } from "../types.ts"; - -const fixturePath = resolve(__dirname, "__fixtures__/wahoo-ride.fit"); -const fitBuffer = readFileSync(fixturePath); - -describe("wahooProvider.convertToGpx", () => { - it("converts a FIT file to valid GPX with correct coordinates", async () => { - const gpx = await wahooProvider.convertToGpx(Buffer.from(fitBuffer)); - - expect(gpx).not.toBeNull(); - expect(gpx).toContain(' { - const gpx = await wahooProvider.convertToGpx(Buffer.from(fitBuffer)); - expect(gpx).not.toBeNull(); - - const latMatches = gpx!.matchAll(/lat="([^"]+)"/g); - const lonMatches = gpx!.matchAll(/lon="([^"]+)"/g); - - const lats = [...latMatches].map((m) => parseFloat(m[1]!)); - const lons = [...lonMatches].map((m) => parseFloat(m[1]!)); - - expect(lats.length).toBeGreaterThan(10); - expect(lons.length).toBeGreaterThan(10); - - for (const lat of lats) { - expect(lat).toBeGreaterThan(-90); - expect(lat).toBeLessThan(90); - // Should be real-world coords, not near-zero from double-conversion - expect(Math.abs(lat)).toBeGreaterThan(1); - } - for (const lon of lons) { - expect(lon).toBeGreaterThan(-180); - expect(lon).toBeLessThan(180); - } - }); - - it("includes ISO 8601 timestamps (not Date objects)", async () => { - const gpx = await wahooProvider.convertToGpx(Buffer.from(fitBuffer)); - expect(gpx).not.toBeNull(); - - const timeMatches = gpx!.matchAll(/