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>
194 lines
6.3 KiB
TypeScript
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 {};
|
|
}
|
|
}
|