trails/apps/journal/app/lib/sync/pushes.server.ts
Ullrich Schäfer a9c0093877
Add Wahoo push action route + OAuth callback resume
Adds /api/sync/push/:provider/:routeId as the user-triggered entry
point for pushing a route. The route action delegates to a new
pushes.server.ts pipeline that:

- resolves the latest route_versions row and uses that GPX (not
  routes.gpx) so the bytes Wahoo gets match the snapshot the user
  sees
- short-circuits on (user, route, version, provider) idempotency:
  a successful prior push returns the existing remote_id without
  re-calling Wahoo
- detects scope_missing before hitting Wahoo and redirects through
  getAuthUrl with a base64url-encoded state carrying pushAfter +
  returnTo
- refreshes tokens once on PushError({ code: "token_expired" }) and
  retries the push, then updates sync_connections in place
- records every outcome in sync_pushes (insert on first attempt,
  update on retry) so the UI can show success/failure state

The OAuth callback handler now decodes the state, resumes a
pushAfter pipeline server-side after exchangeCode, and handles the
?error=access_denied path with a needs_permission notice.

Also flips packages/fit/src/fitsdk.d.ts to a regular .ts side-effect
shim so journal's tsc picks up the @garmin/fitsdk module declaration
when consuming the workspace package via source.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-04-30 22:41:24 +02:00

194 lines
6.3 KiB
TypeScript

import { randomUUID } from "node:crypto";
import { and, desc, eq } from "drizzle-orm";
import { gpxToFitCourse } from "@trails-cool/fit";
import { parseGpxAsync } from "@trails-cool/gpx";
import { routeVersions, syncPushes } from "@trails-cool/db/schema/journal";
import { getDb } from "../db.ts";
import { getRoute } from "../routes.server.ts";
import { getConnection, updateTokens } from "./connections.server.ts";
import { getProvider } from "./registry.ts";
import { PushError, providerSupportsPush } from "./types.ts";
import type { TokenSet } from "./types.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: "error"; code: "validation" | "rate_limit" | "token_expired" | "generic"; message: string };
export interface PushRouteOptions {
userId: string;
providerId: string;
routeId: string;
}
/**
* End-to-end push pipeline. Resolves the route, checks ownership and scopes,
* short-circuits on already-pushed versions, refreshes tokens on 401, and
* records the outcome in `sync_pushes`.
*/
export async function pushRouteToProvider(opts: PushRouteOptions): Promise<PushOutcome> {
const { userId, providerId, routeId } = opts;
const db = getDb();
const provider = getProvider(providerId);
if (!provider) return { status: "not_found" };
if (!providerSupportsPush(provider)) return { status: "unsupported_provider" };
const route = await getRoute(routeId);
if (!route) return { status: "not_found" };
if (route.ownerId !== userId) return { status: "not_owner" };
const connection = await getConnection(userId, providerId);
if (!connection) return { status: "no_connection" };
if (!connection.grantedScopes.includes("routes_write")) return { status: "scope_missing" };
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" };
// Idempotency: if this exact (user, route, version) was already pushed, return that result.
const existing = await db
.select()
.from(syncPushes)
.where(
and(
eq(syncPushes.userId, userId),
eq(syncPushes.routeId, routeId),
eq(syncPushes.routeVersion, versionNumber),
eq(syncPushes.provider, providerId),
),
)
.limit(1);
const existingRow = existing[0];
if (existingRow?.pushedAt && existingRow.remoteId) {
return { status: "success", remoteId: existingRow.remoteId, pushedAt: existingRow.pushedAt };
}
// Build payload from the locked-in version GPX, not routes.gpx.
const parsed = await parseGpxAsync(versionGpx);
const points = parsed.tracks.flat();
if (points.length === 0) return { status: "no_geometry" };
const fit = await gpxToFitCourse({
gpx: versionGpx,
name: route.name,
description: route.description ?? undefined,
});
const externalId = `route:${routeId}:v${versionNumber}`;
const payload = {
fit,
externalId,
providerUpdatedAt: new Date(),
name: route.name,
description: route.description ?? undefined,
startLat: points[0]!.lat,
startLng: points[0]!.lon,
distance: parsed.distance,
ascent: parsed.elevation.gain,
};
let tokens: TokenSet = {
accessToken: connection.accessToken,
refreshToken: connection.refreshToken,
expiresAt: connection.expiresAt,
providerUserId: connection.providerUserId ?? undefined,
};
if (Date.now() >= tokens.expiresAt.getTime()) {
tokens = await provider.refreshToken(tokens.refreshToken);
await updateTokens(connection.id, tokens);
}
const recordResult = async (
success: boolean,
remoteId: string | null,
error: string | null,
): Promise<void> => {
const now = new Date();
if (existingRow) {
await db
.update(syncPushes)
.set({
remoteId,
pushedAt: success ? now : null,
error,
updatedAt: now,
})
.where(eq(syncPushes.id, existingRow.id));
} else {
await db.insert(syncPushes).values({
id: randomUUID(),
userId,
routeId,
routeVersion: versionNumber,
provider: providerId,
externalId,
remoteId,
pushedAt: success ? now : null,
error,
});
}
};
try {
let result;
try {
result = await provider.pushRoute(tokens, payload);
} catch (e) {
if (e instanceof PushError && e.code === "token_expired") {
// 401 mid-call — refresh once and retry.
tokens = await provider.refreshToken(tokens.refreshToken);
await updateTokens(connection.id, tokens);
result = await provider.pushRoute(tokens, payload);
} else {
throw e;
}
}
await recordResult(true, result.remoteId, null);
return { status: "success", remoteId: result.remoteId, pushedAt: new Date() };
} catch (e) {
if (e instanceof PushError) {
await recordResult(false, null, `${e.code}: ${e.message}`);
if (e.code === "scope_missing") return { status: "scope_missing" };
return { status: "error", code: e.code, message: e.message };
}
const message = e instanceof Error ? e.message : String(e);
await recordResult(false, null, `generic: ${message}`);
return { status: "error", code: "generic", message };
}
}
// State payload encoded into the OAuth `state` query param so the callback
// can resume a push that was interrupted by a re-auth redirect.
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 {};
}
}