Merge pull request #377 from trails-cool/stigi/arch-deepening-3-1-6

Deepen three architectural seams: FIT consolidation, host election, injectable db

apps/journal/app/lib/connected-services/fit.ts

@@ -0,0 +1,38 @@
+// Shared FIT file → GPX converter for provider importers.
+//
+// FIT is an open standard (Garmin/ANT+) used by Wahoo, Garmin, Coros, and
+// others. This module is shared across all providers that produce FIT files
+// so the conversion logic lives in one place.
+
+import FitParser from "fit-file-parser";
+import { generateGpx } from "@trails-cool/gpx";
+
+export async function fitToGpx(buffer: Buffer, name: string): Promise<string | null> {
+  const parsed = await new Promise<Record<string, unknown>>((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] });
+}

apps/journal/app/lib/connected-services/providers/wahoo/importer.ts

@@ -4,10 +4,8 @@
 // Credentials always flow through ctx.withFreshCredentials — this module
 // never reads the connected_services credentials JSONB directly.
 
-import FitParser from "fit-file-parser";
+import { fitToGpx } from "../../fit.ts";
-import { generateGpx } from "@trails-cool/gpx";
+import { importActivity, isAlreadyImported } from "../../../sync/imports.server.ts";
-import { createActivity } from "../../../activities.server.ts";
-import { recordImport, isAlreadyImported } from "../../../sync/imports.server.ts";
 import type {
   CapabilityContext,
   ImportableList,
@@ -69,37 +67,6 @@ function toImportable(w: WahooWorkout) {
   };
 }
 
-async function fitToGpx(buffer: Buffer, name: string): Promise<string | null> {
-  const parsed = await new Promise<Record<string, unknown>>((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<Buffer> {
   // Wahoo CDN URLs are pre-signed; no auth header needed (and adding one
   // breaks them).
@@ -163,11 +130,10 @@ export const wahooImporter: Importer = {
       gpx = await fitToGpx(buffer, workout.name || "Wahoo workout");
     }
 
-    const activityId = await createActivity(userId, {
+    const { activityId } = await importActivity(userId, "wahoo", workoutId, {
       name: workout.name || `Wahoo workout ${workoutId}`,
       gpx: gpx ?? undefined,
     });
-    await recordImport(userId, "wahoo", workoutId, activityId);
 
     return { activityId, hadGeometry: gpx !== null };
   },

apps/journal/app/lib/connected-services/providers/wahoo/webhook.test.ts

@@ -6,18 +6,14 @@
 import { describe, it, expect, beforeEach, vi } from "vitest";
 
 const fetchSpy = vi.fn();
-const mockCreateActivity = vi.fn();
+const mockImportActivity = 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,
+  importActivity: mockImportActivity,
 }));
 vi.mock("../../manager.ts", () => ({
   getServiceByProviderUser: mockGetServiceByProviderUser,
@@ -27,9 +23,8 @@ vi.mock("../../manager.ts", () => ({
 beforeEach(() => {
   fetchSpy.mockReset();
   globalThis.fetch = fetchSpy as unknown as typeof fetch;
-  mockCreateActivity.mockReset();
+  mockImportActivity.mockReset();
   mockIsAlreadyImported.mockReset();
-  mockRecordImport.mockReset();
   mockGetServiceByProviderUser.mockReset();
   mockWithFreshCredentials.mockReset();
 });
@@ -73,18 +68,7 @@ describe("wahooWebhook.handle", () => {
       provider: "wahoo",
     });
     mockIsAlreadyImported.mockResolvedValue(false);
-    mockCreateActivity.mockResolvedValue("act-1");
+    mockImportActivity.mockResolvedValue({ activityId: "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<unknown>) =>
-        fn({
-          access_token: "a",
-          refresh_token: "r",
-          expires_at: new Date(Date.now() + 3600_000).toISOString(),
-        }),
-    );
 
     await wahooWebhook.handle({
       eventType: "workout_summary",
@@ -93,11 +77,12 @@ describe("wahooWebhook.handle", () => {
       // no fileUrl — the activity is created without GPX
     });
 
-    expect(mockCreateActivity).toHaveBeenCalledWith(
+    expect(mockImportActivity).toHaveBeenCalledWith(
       "u1",
+      "wahoo",
+      "42",
       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 () => {
@@ -109,8 +94,7 @@ describe("wahooWebhook.handle", () => {
       workoutId: "42",
     });
 
-    expect(mockCreateActivity).not.toHaveBeenCalled();
+    expect(mockImportActivity).not.toHaveBeenCalled();
-    expect(mockRecordImport).not.toHaveBeenCalled();
   });
 
   it("silently skips when the workout was already imported (idempotency)", async () => {
@@ -127,7 +111,6 @@ describe("wahooWebhook.handle", () => {
       workoutId: "42",
     });
 
-    expect(mockCreateActivity).not.toHaveBeenCalled();
+    expect(mockImportActivity).not.toHaveBeenCalled();
-    expect(mockRecordImport).not.toHaveBeenCalled();
   });
 });

apps/journal/app/lib/connected-services/providers/wahoo/webhook.ts

@@ -5,10 +5,8 @@
 // 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 { fitToGpx } from "../../fit.ts";
-import { generateGpx } from "@trails-cool/gpx";
+import { isAlreadyImported, importActivity } from "../../../sync/imports.server.ts";
-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";
@@ -23,35 +21,6 @@ interface WahooWebhookBody {
   };
 }
 
-async function fitToGpx(buffer: Buffer): Promise<string | null> {
-  const parsed = await new Promise<Record<string, unknown>>((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 {
@@ -88,13 +57,12 @@ export const wahooWebhook: WebhookReceiver = {
         if (!resp.ok) throw new Error(`Wahoo file download failed: ${resp.status}`);
         return Buffer.from(await resp.arrayBuffer());
       });
-      gpx = await fitToGpx(buffer);
+      gpx = await fitToGpx(buffer, "Wahoo workout");
     }
 
-    const activityId = await createActivity(service.userId, {
+    await importActivity(service.userId, "wahoo", event.workoutId, {
-      name: `Wahoo workout`,
+      name: "Wahoo workout",
       gpx: gpx ?? undefined,
     });
-    await recordImport(service.userId, "wahoo", event.workoutId, activityId);
   },
 };

apps/journal/app/lib/db.ts

@@ -8,3 +8,7 @@ export function getDb(): Database {
   }
   return _db;
 }
+
+export function setDb(db: Database | null): void {
+  _db = db;
+}

apps/journal/app/lib/sync/imports.server.ts

@@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto";
 import { eq, and, inArray } from "drizzle-orm";
 import { getDb } from "../db.ts";
 import { syncImports } from "@trails-cool/db/schema/journal";
+import { createActivity } from "../activities.server.ts";
 
 export async function recordImport(
   userId: string,
@@ -43,6 +44,20 @@ export async function deleteImportByActivity(activityId: string) {
   await db.delete(syncImports).where(eq(syncImports.activityId, activityId));
 }
 
+// Single callsite for "create an activity from an imported workout and record
+// the dedup entry". Providers call this instead of calling createActivity +
+// recordImport separately, so the pair stays atomic from the provider's view.
+export async function importActivity(
+  userId: string,
+  provider: string,
+  externalWorkoutId: string,
+  input: { name: string; gpx?: string },
+): Promise<{ activityId: string }> {
+  const activityId = await createActivity(userId, { name: input.name, gpx: input.gpx });
+  await recordImport(userId, provider, externalWorkoutId, activityId);
+  return { activityId };
+}
+
 export async function getImportedIds(
   userId: string,
   provider: string,

apps/planner/app/lib/use-host-election.ts

@@ -0,0 +1,28 @@
+import { useEffect, useState } from "react";
+import { electHost } from "./host-election.ts";
+import type { YjsState } from "./use-yjs.ts";
+
+export function useHostElection(yjs: YjsState | null): boolean {
+  const [isHost, setIsHost] = useState(false);
+
+  useEffect(() => {
+    if (!yjs) return;
+
+    const checkHost = () => {
+      const states = yjs.awareness.getStates() as Map<number, Record<string, unknown>>;
+      const localId = yjs.awareness.clientID;
+      const { isHost: amHost, role } = electHost(states, localId);
+      setIsHost(amHost);
+      yjs.awareness.setLocalStateField("role", role);
+    };
+
+    yjs.awareness.on("change", checkHost);
+    checkHost();
+
+    return () => {
+      yjs.awareness.off("change", checkHost);
+    };
+  }, [yjs]);
+
+  return isHost;
+}

apps/planner/app/lib/use-routing.ts

@@ -1,7 +1,7 @@
 import { useCallback, useEffect, useRef, useState } from "react";
 import * as Y from "yjs";
 import type { YjsState } from "./use-yjs.ts";
-import { electHost } from "./host-election.ts";
+import { useHostElection } from "./use-host-election.ts";
 import {
   hashNoGoAreas,
   mergeGeoJsonSegments,
@@ -51,7 +51,7 @@ function restoreWaypoints(yjs: YjsState, snapshot: WaypointData[], restoringRef:
 export type RouteError = "no_route" | "failed" | "rate_limit" | null;
 
 export function useRouting(yjs: YjsState | null, sessionId: string) {
-  const [isHost, setIsHost] = useState(false);
+  const isHost = useHostElection(yjs);
   const [computing, setComputing] = useState(false);
   const [routeError, setRouteError] = useState<RouteError>(null);
   const [routeStats, setRouteStats] = useState<RouteStats>({});
@@ -67,26 +67,6 @@ export function useRouting(yjs: YjsState | null, sessionId: string) {
   // instead of N-1 on every recompute.
   const segmentCacheRef = useRef<SegmentCache>(new SegmentCache());
 
-  // Host election via Yjs awareness
-  useEffect(() => {
-    if (!yjs) return;
-
-    const checkHost = () => {
-      const states = yjs.awareness.getStates() as Map<number, Record<string, unknown>>;
-      const localId = yjs.awareness.clientID;
-      const { isHost: amHost, role } = electHost(states, localId);
-      setIsHost(amHost);
-      yjs.awareness.setLocalStateField("role", role);
-    };
-
-    yjs.awareness.on("change", checkHost);
-    checkHost();
-
-    return () => {
-      yjs.awareness.off("change", checkHost);
-    };
-  }, [yjs]);
-
   const computeRoute = useCallback(
     async (waypoints: WaypointData[]) => {
       if (!yjs || !isHost || waypoints.length < 2) return;