diff --git a/apps/journal/app/jobs/komoot-bulk-import.ts b/apps/journal/app/jobs/komoot-bulk-import.ts new file mode 100644 index 0000000..84c4bc8 --- /dev/null +++ b/apps/journal/app/jobs/komoot-bulk-import.ts @@ -0,0 +1,27 @@ +import type { JobDefinition } from "@trails-cool/jobs"; +import { logger } from "../lib/logger.server.ts"; +import { runKomootBulkImport } from "../lib/komoot-bulk-import.server.ts"; + +type KomootCreds = + | { mode: "public"; komootUserId: string } + | { mode: "authenticated"; email: string; encryptedPassword: string; komootUserId: string }; + +interface KomootBulkImportData extends Record { + batchId: string; + userId: string; + creds: KomootCreds; +} + +export const komootBulkImportJob: JobDefinition = { + name: "komoot-bulk-import", + retryLimit: 1, + expireInSeconds: 1800, + async handler(jobs) { + const batch = Array.isArray(jobs) ? jobs : [jobs]; + for (const job of batch) { + const { batchId, userId, creds } = job.data; + logger.info({ batchId, userId }, "komoot bulk import job started"); + await runKomootBulkImport(batchId, userId, creds); + } + }, +}; diff --git a/apps/journal/app/lib/komoot-bulk-import.server.ts b/apps/journal/app/lib/komoot-bulk-import.server.ts new file mode 100644 index 0000000..7468fd7 --- /dev/null +++ b/apps/journal/app/lib/komoot-bulk-import.server.ts @@ -0,0 +1,114 @@ +import { eq } from "drizzle-orm"; +import { getDb } from "./db.ts"; +import { importBatches } from "@trails-cool/db/schema/journal"; +import { fetchKomootTours, fetchKomootTourGpx } from "./komoot.server.ts"; +import { isAlreadyImported, recordImport } from "./sync/imports.server.ts"; +import { createActivity } from "./activities.server.ts"; +import { decrypt } from "./crypto.server.ts"; +import { logger } from "./logger.server.ts"; + +type KomootCreds = + | { mode: "public"; komootUserId: string } + | { mode: "authenticated"; email: string; encryptedPassword: string; komootUserId: string }; + +function getBasicAuthToken(creds: KomootCreds): string | undefined { + if (creds.mode !== "authenticated") return undefined; + const password = decrypt(creds.encryptedPassword); + return Buffer.from(`${creds.email}:${password}`).toString("base64"); +} + +export async function runKomootBulkImport( + batchId: string, + userId: string, + creds: KomootCreds, +): Promise { + const db = getDb(); + const basicAuthToken = getBasicAuthToken(creds); + const komootUserId = creds.komootUserId; + + await db + .update(importBatches) + .set({ status: "running" }) + .where(eq(importBatches.id, batchId)); + + try { + let page = 1; + let hasMore = true; + let totalFound = 0; + let importedCount = 0; + let duplicateCount = 0; + + while (hasMore) { + let result: Awaited>; + try { + result = await fetchKomootTours(komootUserId, page, basicAuthToken); + } catch (err) { + logger.warn({ batchId, page, err }, "komoot bulk import: fetchKomootTours failed, stopping"); + break; + } + + totalFound += result.tours.length; + hasMore = page < result.totalPages; + page++; + + await db + .update(importBatches) + .set({ totalFound }) + .where(eq(importBatches.id, batchId)); + + for (const tour of result.tours) { + if (await isAlreadyImported(userId, "komoot", tour.id)) { + duplicateCount++; + await db + .update(importBatches) + .set({ duplicateCount }) + .where(eq(importBatches.id, batchId)); + continue; + } + + let gpx: string | undefined; + try { + gpx = await fetchKomootTourGpx(tour.id, basicAuthToken); + } catch { + // No GPX — import metadata only + } + + try { + const activityId = await createActivity(userId, { + name: tour.name, + gpx, + distance: tour.distance > 0 ? tour.distance : null, + duration: tour.duration > 0 ? tour.duration : null, + startedAt: tour.date ? new Date(tour.date) : null, + }); + await recordImport(userId, "komoot", tour.id, activityId); + importedCount++; + } catch (err) { + logger.warn({ batchId, tourId: tour.id, err }, "komoot bulk import: failed to import tour"); + } + + await db + .update(importBatches) + .set({ importedCount, duplicateCount }) + .where(eq(importBatches.id, batchId)); + } + } + + await db + .update(importBatches) + .set({ status: "completed", totalFound, importedCount, duplicateCount, completedAt: new Date() }) + .where(eq(importBatches.id, batchId)); + + logger.info({ batchId, totalFound, importedCount, duplicateCount }, "komoot bulk import completed"); + } catch (err) { + await db + .update(importBatches) + .set({ + status: "failed", + errorMessage: err instanceof Error ? err.message : String(err), + completedAt: new Date(), + }) + .where(eq(importBatches.id, batchId)); + throw err; + } +} diff --git a/apps/journal/app/routes.ts b/apps/journal/app/routes.ts index 1fe92e5..f28fbda 100644 --- a/apps/journal/app/routes.ts +++ b/apps/journal/app/routes.ts @@ -55,6 +55,8 @@ export default [ route("sync/import/:provider", "routes/sync.import.$provider.tsx"), route("api/sync/komoot/verify", "routes/api.sync.komoot.verify.ts"), route("api/sync/komoot/connect", "routes/api.sync.komoot.connect.ts"), + route("api/sync/komoot/import", "routes/api.sync.komoot.import.ts"), + route("api/sync/komoot/import-status", "routes/api.sync.komoot.import-status.ts"), route("api/sync/connect/:provider", "routes/api.sync.connect.$provider.ts"), route("api/sync/callback/:provider", "routes/api.sync.callback.$provider.ts"), route("api/sync/disconnect/:provider", "routes/api.sync.disconnect.$provider.ts"), diff --git a/apps/journal/app/routes/api.sync.komoot.import-status.ts b/apps/journal/app/routes/api.sync.komoot.import-status.ts new file mode 100644 index 0000000..c0a507f --- /dev/null +++ b/apps/journal/app/routes/api.sync.komoot.import-status.ts @@ -0,0 +1,41 @@ +// GET /api/sync/komoot/import-status +// Returns the most recent import batch for the authenticated user's Komoot connection. + +import { data, redirect } from "react-router"; +import { desc, eq, and } from "drizzle-orm"; +import type { Route } from "./+types/api.sync.komoot.import-status"; +import { getSessionUser } from "~/lib/auth/session.server"; +import { getService } from "~/lib/connected-services/manager"; +import { getDb } from "~/lib/db"; +import { importBatches } from "@trails-cool/db/schema/journal"; + +export async function loader({ request }: Route.LoaderArgs) { + const user = await getSessionUser(request); + if (!user) return redirect("/auth/login"); + + const service = await getService(user.id, "komoot"); + if (!service) return data({ batch: null }); + + const db = getDb(); + const [batch] = await db + .select() + .from(importBatches) + .where(and(eq(importBatches.userId, user.id), eq(importBatches.connectionId, service.id))) + .orderBy(desc(importBatches.startedAt)) + .limit(1); + + return data({ + batch: batch + ? { + id: batch.id, + status: batch.status, + totalFound: batch.totalFound, + importedCount: batch.importedCount, + duplicateCount: batch.duplicateCount, + errorMessage: batch.errorMessage, + startedAt: batch.startedAt.toISOString(), + completedAt: batch.completedAt?.toISOString() ?? null, + } + : null, + }); +} diff --git a/apps/journal/app/routes/api.sync.komoot.import.ts b/apps/journal/app/routes/api.sync.komoot.import.ts new file mode 100644 index 0000000..d0f0267 --- /dev/null +++ b/apps/journal/app/routes/api.sync.komoot.import.ts @@ -0,0 +1,39 @@ +// POST /api/sync/komoot/import +// Enqueues a background bulk import of all Komoot tours for the connected user. +// Returns { batchId } immediately; poll /api/sync/komoot/import-status for progress. + +import { randomUUID } from "node:crypto"; +import { data, redirect } from "react-router"; +import type { Route } from "./+types/api.sync.komoot.import"; +import { getSessionUser } from "~/lib/auth/session.server"; +import { getService } from "~/lib/connected-services/manager"; +import { getBoss } from "~/lib/boss.server"; +import { getDb } from "~/lib/db"; +import { importBatches } from "@trails-cool/db/schema/journal"; + +export async function action({ request }: Route.ActionArgs) { + const user = await getSessionUser(request); + if (!user) return redirect("/auth/login"); + + const service = await getService(user.id, "komoot"); + if (!service) return data({ error: "not_connected" }, { status: 400 }); + + const db = getDb(); + const batchId = randomUUID(); + await db.insert(importBatches).values({ + id: batchId, + userId: user.id, + connectionId: service.id, + provider: "komoot", + status: "pending", + }); + + const boss = getBoss(); + await boss.send("komoot-bulk-import", { + batchId, + userId: user.id, + creds: service.credentials, + }); + + return data({ batchId }); +} diff --git a/apps/journal/app/routes/sync.import.komoot.tsx b/apps/journal/app/routes/sync.import.komoot.tsx index 02bdfdf..07ab181 100644 --- a/apps/journal/app/routes/sync.import.komoot.tsx +++ b/apps/journal/app/routes/sync.import.komoot.tsx @@ -1,28 +1,15 @@ -// Komoot-specific import page. Mirrors the generic sync.import.$provider page -// but fetches GPX directly from Komoot instead of downloading a FIT file. +// Komoot import page — shows live progress for background bulk imports +// and lets the user trigger a new import run. -import { useCallback, useEffect, useRef, useState } from "react"; -import { data, redirect, useFetcher } from "react-router"; +import { useEffect, useRef } from "react"; +import { data, redirect, useFetcher, useRevalidator } from "react-router"; import { useTranslation } from "react-i18next"; import type { Route } from "./+types/sync.import.komoot"; import { getSessionUser } from "~/lib/auth/session.server"; -import { getService, capabilityContextFor } from "~/lib/connected-services"; -import { getManifest } from "~/lib/connected-services"; -import { getImportedIds, recordImport } from "~/lib/sync/imports.server"; -import { createActivity } from "~/lib/activities.server"; -import { fetchKomootTourGpx } from "~/lib/komoot.server"; -import { decrypt } from "~/lib/crypto.server"; -import { ClientDate } from "~/components/ClientDate"; - -type KomootCreds = - | { mode: "public"; komootUserId: string } - | { mode: "authenticated"; email: string; encryptedPassword: string; komootUserId: string }; - -function getBasicAuthToken(creds: KomootCreds): string | undefined { - if (creds.mode !== "authenticated") return undefined; - const password = decrypt(creds.encryptedPassword); - return Buffer.from(`${creds.email}:${password}`).toString("base64"); -} +import { getService } from "~/lib/connected-services"; +import { getDb } from "~/lib/db"; +import { importBatches } from "@trails-cool/db/schema/journal"; +import { desc, eq, and } from "drizzle-orm"; export function meta() { return [{ title: "Import from Komoot — trails.cool" }]; @@ -32,31 +19,30 @@ export async function loader({ request }: Route.LoaderArgs) { const user = await getSessionUser(request); if (!user) return redirect("/auth/login"); - const manifest = getManifest("komoot"); - if (!manifest?.importer) throw data({ error: "Komoot not configured" }, { status: 500 }); - const service = await getService(user.id, "komoot"); if (!service) return redirect("/settings/connections/komoot"); - const url = new URL(request.url); - const page = parseInt(url.searchParams.get("page") ?? "1"); - - const ctx = capabilityContextFor(service.id); - const workoutList = await manifest.importer.listImportable(ctx, page); - - const importedIds = await getImportedIds( - user.id, - "komoot", - workoutList.workouts.map((w) => w.id), - ); + const db = getDb(); + const [batch] = await db + .select() + .from(importBatches) + .where(and(eq(importBatches.userId, user.id), eq(importBatches.connectionId, service.id))) + .orderBy(desc(importBatches.startedAt)) + .limit(1); return data({ - workouts: workoutList.workouts.map((w) => ({ - ...w, - imported: importedIds.has(w.id), - })), - page: workoutList.page, - totalPages: Math.ceil(workoutList.total / workoutList.perPage), + batch: batch + ? { + id: batch.id, + status: batch.status, + totalFound: batch.totalFound, + importedCount: batch.importedCount, + duplicateCount: batch.duplicateCount, + errorMessage: batch.errorMessage, + startedAt: batch.startedAt.toISOString(), + completedAt: batch.completedAt?.toISOString() ?? null, + } + : null, }); } @@ -64,193 +50,159 @@ export async function action({ request }: Route.ActionArgs) { const user = await getSessionUser(request); if (!user) return redirect("/auth/login"); - const service = await getService(user.id, "komoot"); - if (!service) return redirect("/settings/connections/komoot"); - - const formData = await request.formData(); - const tourId = formData.get("workoutId") as string; - const workoutName = formData.get("workoutName") as string; - const startedAt = formData.get("startedAt") as string; - const distance = formData.get("distance") as string; - const duration = formData.get("duration") as string; - - const creds = service.credentials as KomootCreds; - const basicAuthToken = getBasicAuthToken(creds); - - let gpx: string | undefined; - try { - gpx = await fetchKomootTourGpx(tourId, basicAuthToken); - } catch { - // No GPX available — import without geometry - } - - const activityId = await createActivity(user.id, { - name: workoutName || `Komoot tour ${tourId}`, - gpx, - distance: distance ? parseFloat(distance) : null, - duration: duration ? parseInt(duration) : null, - startedAt: startedAt ? new Date(startedAt) : null, - }); - - await recordImport(user.id, "komoot", tourId, activityId); - - return data({ imported: tourId }); -} - -function TourRow({ - workout, - alreadyImported, -}: { - workout: { id: string; name: string; startedAt: string; distance: number | null; duration: number | null }; - alreadyImported: boolean; -}) { - const { t } = useTranslation("journal"); - const fetcher = useFetcher<{ imported?: string }>(); - const isImporting = fetcher.state !== "idle"; - const isImported = alreadyImported || fetcher.data?.imported === workout.id; - - return ( -
  • -
    -

    {workout.name}

    -
    - {workout.startedAt && } - {workout.distance != null && {(workout.distance / 1000).toFixed(1)} km} - {workout.duration != null && {Math.round(workout.duration / 60)} min} -
    -
    - {isImported ? ( - {t("sync.imported")} - ) : isImporting ? ( - {t("sync.importing")} - ) : ( - - - - - - - - - )} -
  • + // Delegate to the API route — just redirect so the page reloads with + // the new batch after the POST. + const resp = await fetch( + new URL("/api/sync/komoot/import", new URL(request.url).origin), + { method: "POST", headers: { cookie: request.headers.get("cookie") ?? "" } }, ); + if (!resp.ok) { + const body = (await resp.json()) as { error?: string }; + return data({ error: body.error ?? "failed" }, { status: resp.status }); + } + return redirect("/sync/import/komoot"); } -function tourFormData(w: { id: string; name: string; startedAt: string; distance: number | null; duration: number | null }) { - const fd = new FormData(); - fd.set("workoutId", w.id); - fd.set("workoutName", w.name); - fd.set("startedAt", w.startedAt ?? ""); - fd.set("distance", w.distance != null ? String(w.distance) : ""); - fd.set("duration", w.duration != null ? String(w.duration) : ""); - return fd; +function formatDuration(seconds: number): string { + const h = Math.floor(seconds / 3600); + const m = Math.floor((seconds % 3600) / 60); + if (h > 0) return `${h}h ${m}m`; + return `${m}m`; } export default function KomootImportPage({ loaderData }: Route.ComponentProps) { - const { workouts, page, totalPages } = loaderData; + const { batch } = loaderData; const { t } = useTranslation("journal"); + const revalidator = useRevalidator(); + const triggerFetcher = useFetcher<{ error?: string }>(); + const pollingRef = useRef | null>(null); - const importableWorkouts = workouts.filter((w) => !w.imported); - const [importAllActive, setImportAllActive] = useState(false); - const [importAllIndex, setImportAllIndex] = useState(0); - const importAllFetcher = useFetcher<{ imported?: string }>(); - const importAllRef = useRef({ index: 0, workouts: importableWorkouts }); + const isActive = batch?.status === "pending" || batch?.status === "running"; useEffect(() => { - if (!importAllActive) return; - if (importAllFetcher.state !== "idle") return; - - const nextIndex = importAllFetcher.data - ? importAllRef.current.index + 1 - : importAllRef.current.index; - importAllRef.current.index = nextIndex; - setImportAllIndex(nextIndex); - - if (nextIndex >= importAllRef.current.workouts.length) { - setImportAllActive(false); - return; + if (isActive) { + pollingRef.current = setInterval(() => { + revalidator.revalidate(); + }, 2000); } + return () => { + if (pollingRef.current) clearInterval(pollingRef.current); + }; + }, [isActive]); - const w = importAllRef.current.workouts[nextIndex]!; - importAllFetcher.submit(tourFormData(w), { method: "post" }); - }, [importAllActive, importAllFetcher.state, importAllFetcher.data]); - - const startImportAll = useCallback(() => { - if (importableWorkouts.length === 0) return; - importAllRef.current = { index: 0, workouts: importableWorkouts }; - setImportAllIndex(0); - setImportAllActive(true); - importAllFetcher.submit(tourFormData(importableWorkouts[0]!), { method: "post" }); - }, [importableWorkouts, importAllFetcher]); + const elapsedSeconds = batch + ? Math.floor( + (batch.completedAt + ? new Date(batch.completedAt).getTime() + : Date.now()) - new Date(batch.startedAt).getTime(), + ) / 1000 + : 0; return ( -
    -
    -

    - {t("sync.importFrom", { provider: "Komoot" })} -

    - {importableWorkouts.length > 0 && ( - importAllActive ? ( - - {t("sync.importingProgress", { - current: importAllIndex + 1, - total: importAllRef.current.workouts.length, - })} - - ) : ( - - ) +
    +

    + {t("sync.importFrom", { provider: "Komoot" })} +

    + +
    + {!batch && ( +
    +

    {t("komoot.import.noImportYet")}

    + + + +
    + )} + + {batch && ( +
    +
    + + {(batch.status === "completed" || batch.status === "failed") && ( + + + + )} +
    + + {isActive && ( +
    +
    0 + ? `${Math.round(((batch.importedCount + batch.duplicateCount) / batch.totalFound) * 100)}%` + : "5%", + }} + /> +
    + )} + +
    + + + +
    + + {batch.status === "completed" && ( +

    + {t("komoot.import.completedIn", { duration: formatDuration(elapsedSeconds) })} +

    + )} + + {batch.status === "failed" && batch.errorMessage && ( +

    + {batch.errorMessage} +

    + )} +
    )}
    - {workouts.length === 0 ? ( -

    {t("sync.noWorkouts")}

    - ) : ( -
      - {workouts.map((w) => ( - - ))} -
    - )} - - {totalPages > 1 && ( -
    - {page > 1 && ( - - {t("sync.previous")} - - )} - - {page} / {totalPages} - - {page < totalPages && ( - - {t("sync.next")} - - )} -
    + {batch?.status === "completed" && ( +

    + + {t("komoot.import.viewActivities")} + +

    )}
    ); } + +function StatusBadge({ status, t }: { status: string; t: (k: string) => string }) { + const styles: Record = { + pending: "bg-gray-100 text-gray-600", + running: "bg-blue-100 text-blue-700", + completed: "bg-green-100 text-green-700", + failed: "bg-red-100 text-red-700", + }; + return ( + + {t(`komoot.import.status.${status}`)} + + ); +} + +function StatBox({ label, value }: { label: string; value: number }) { + return ( +
    +

    {value}

    +

    {label}

    +
    + ); +} diff --git a/apps/journal/server.ts b/apps/journal/server.ts index 6367191..d77a5c9 100644 --- a/apps/journal/server.ts +++ b/apps/journal/server.ts @@ -143,7 +143,9 @@ server.listen(port, async () => { // notifications wired up would silently drop fan-out enqueues. const { notificationsFanoutJob } = await import("./app/jobs/notifications-fanout.ts"); const { notificationsPurgeJob } = await import("./app/jobs/notifications-purge.ts"); - jobs.push(notificationsFanoutJob, notificationsPurgeJob); + const { komootBulkImportJob } = await import("./app/jobs/komoot-bulk-import.ts"); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob as any); const boss = createBoss(process.env.DATABASE_URL ?? "postgres://trails:trails@localhost:5432/trails"); await startWorker(boss, jobs); diff --git a/packages/db/src/schema/journal.ts b/packages/db/src/schema/journal.ts index d6074e4..a21a88f 100644 --- a/packages/db/src/schema/journal.ts +++ b/packages/db/src/schema/journal.ts @@ -276,6 +276,26 @@ export const syncPushes = journalSchema.table( // row in this change. `acceptedAt` is always set today (auto-accept for // public local profiles); the column stays nullable so federation's Pending // state lands cleanly. +export type ImportBatchStatus = "pending" | "running" | "completed" | "failed"; + +export const importBatches = journalSchema.table("import_batches", { + id: text("id").primaryKey(), + userId: text("user_id") + .notNull() + .references(() => users.id, { onDelete: "cascade" }), + connectionId: text("connection_id") + .notNull() + .references(() => connectedServices.id, { onDelete: "cascade" }), + provider: text("provider").notNull(), + status: text("status").$type().notNull().default("pending"), + totalFound: integer("total_found").notNull().default(0), + importedCount: integer("imported_count").notNull().default(0), + duplicateCount: integer("duplicate_count").notNull().default(0), + errorMessage: text("error_message"), + startedAt: timestamp("started_at", { withTimezone: true }).notNull().defaultNow(), + completedAt: timestamp("completed_at", { withTimezone: true }), +}); + export const follows = journalSchema.table("follows", { id: text("id").primaryKey(), followerId: text("follower_id") diff --git a/packages/i18n/src/locales/de.ts b/packages/i18n/src/locales/de.ts index ee60ca2..b5f5714 100644 --- a/packages/i18n/src/locales/de.ts +++ b/packages/i18n/src/locales/de.ts @@ -445,6 +445,23 @@ export default { connectSuccess: "Komoot-Konto verbunden.", authError: "Ungültige Komoot-Zugangsdaten. Bitte E-Mail und Passwort prüfen.", modeBadge: "Modus: {{mode}}", + import: { + noImportYet: "Es wurde noch kein Import durchgeführt. Klicke unten, um alle deine Komoot-Touren zu importieren.", + startImport: "Alle Touren importieren", + starting: "Wird gestartet…", + runAgain: "Erneut ausführen", + found: "Gefunden", + imported: "Importiert", + skipped: "Bereits importiert", + completedIn: "Abgeschlossen in {{duration}}", + viewActivities: "Aktivitäten ansehen", + status: { + pending: "In Warteschlange", + running: "Wird importiert…", + completed: "Abgeschlossen", + failed: "Fehlgeschlagen", + }, + }, }, sync: { import: "Importieren", diff --git a/packages/i18n/src/locales/en.ts b/packages/i18n/src/locales/en.ts index 426e177..e6c5770 100644 --- a/packages/i18n/src/locales/en.ts +++ b/packages/i18n/src/locales/en.ts @@ -445,6 +445,23 @@ export default { connectSuccess: "Komoot account connected.", authError: "Invalid Komoot credentials. Please check your email and password.", modeBadge: "Mode: {{mode}}", + import: { + noImportYet: "No import has been run yet. Click the button below to import all your Komoot tours.", + startImport: "Import all tours", + starting: "Starting…", + runAgain: "Run again", + found: "Found", + imported: "Imported", + skipped: "Already imported", + completedIn: "Completed in {{duration}}", + viewActivities: "View your activities", + status: { + pending: "Queued", + running: "Importing…", + completed: "Completed", + failed: "Failed", + }, + }, }, sync: { import: "Import",