Webhooks

{
  "version": 1,
  "id": "33333333-3333-4333-a333-333333333333",
  "type": "activity.updated",
  "occurred_at": "2026-10-11T08:41:27.115Z",
  "data": {
    "grant_id": "22222222-2222-4222-8222-222222222222",
    "connection_version": "1",
    "scopes": ["daily_steps:read"],
    "activity_types": ["steps"],
    "from": "2026-10-11",
    "to": "2026-10-11"
  }
}

{
  "version": 1,
  "id": "55555555-5555-4555-a555-555555555555",
  "type": "connection.authorized",
  "occurred_at": "2026-10-10T09:15:02.481Z",
  "data": {
    "grant_id": "22222222-2222-4222-8222-222222222222",
    "connection_version": "1",
    "scopes": ["daily_steps:read", "cycling_daily:read"]
  }
}

{
  "version": 1,
  "id": "66666666-6666-4666-a666-666666666666",
  "type": "connection.revoked",
  "occurred_at": "2026-11-02T17:20:45.009Z",
  "data": {
    "grant_id": "22222222-2222-4222-8222-222222222222",
    "connection_version": "1"
  }
}

signed_payload = Walk15-Timestamp + "\n" + Walk15-Event-Id + "\n" + Walk15-Delivery-Id + "\n" + raw_body
Walk15-Signature = "v1=" + lowercase_hex(HMAC_SHA256(signing_secret, signed_payload))

import { createHmac, timingSafeEqual } from "node:crypto";
import { createServer, type IncomingHttpHeaders } from "node:http";

export type Walk15Event = {
  version: 1;
  id: string;
  type: "connection.authorized" | "connection.revoked" | "activity.updated";
  occurred_at: string;
  data: {
    grant_id: string;
    connection_version: string;
    scopes?: string[];
    activity_types?: ("steps" | "cycling")[];
    from?: string;
    to?: string;
  };
};

export interface Inbox {
  /** Durably store the event; a duplicate id must be a no-op (INSERT … ON CONFLICT DO NOTHING). */
  insertOnce(event: Walk15Event): Promise<void>;
}

const header = (headers: IncomingHttpHeaders, name: string): string | undefined => {
  const value = headers[name];
  return typeof value === "string" ? value : undefined;
};

export const verifyWalk15Signature = (
  secret: string,
  headers: IncomingHttpHeaders,
  rawBody: Buffer,
  nowMs: number = Date.now(),
): boolean => {
  const timestamp = header(headers, "walk15-timestamp");
  const eventId = header(headers, "walk15-event-id");
  const deliveryId = header(headers, "walk15-delivery-id");
  const signature = header(headers, "walk15-signature");
  if (!timestamp || !eventId || !deliveryId || !signature || !/^\d{10}$/.test(timestamp)) return false;
  if (Math.abs(nowMs / 1000 - Number(timestamp)) > 300) return false;
  const expected =
    "v1=" +
    createHmac("sha256", secret).update(`${timestamp}\n${eventId}\n${deliveryId}\n`).update(rawBody).digest("hex");
  const actual = Buffer.from(signature);
  return actual.length === expected.length && timingSafeEqual(actual, Buffer.from(expected));
};

const EVENT_TYPES = ["connection.authorized", "connection.revoked", "activity.updated"];
const DATE = /^\d{4}-\d{2}-\d{2}$/;

export const parseWalk15Event = (rawBody: Buffer, eventIdHeader: string | undefined): Walk15Event | null => {
  try {
    const event = JSON.parse(rawBody.toString("utf8"));
    const data = event?.data;
    if (event?.version !== 1 || event.id !== eventIdHeader || !EVENT_TYPES.includes(event.type)) return null;
    if (typeof data?.grant_id !== "string" || !/^[1-9]\d*$/.test(String(data.connection_version))) return null;
    if (event.type === "activity.updated" && !(DATE.test(data.from) && DATE.test(data.to) && Array.isArray(data.scopes)))
      return null;
    return event as Walk15Event;
  } catch {
    return null;
  }
};

/** `getSecret` maps Walk15-Key-Version to a signing secret you hold. */
export const createWalk15WebhookServer = (getSecret: (keyVersion: string) => string | undefined, inbox: Inbox) =>
  createServer(async (request, response) => {
    if (request.method !== "POST" || request.url !== "/webhooks/walk15") return void response.writeHead(404).end();

    const chunks: Buffer[] = [];
    let size = 0;
    for await (const chunk of request) {
      size += chunk.length;
      if (size > 65_536) return void response.writeHead(413).end();
      chunks.push(chunk);
    }
    const rawBody = Buffer.concat(chunks);

    const secret = getSecret(header(request.headers, "walk15-key-version") ?? "");
    // Unknown key version (mid-rotation): 503 makes Walk15 retry after you install the new secret.
    if (!secret) return void response.writeHead(503).end();
    if (!verifyWalk15Signature(secret, request.headers, rawBody)) return void response.writeHead(401).end();

    const event = parseWalk15Event(rawBody, header(request.headers, "walk15-event-id"));
    if (!event) return void response.writeHead(400).end();

    try {
      await inbox.insertOnce(event);
    } catch {
      return void response.writeHead(503).end(); // not stored: let Walk15 retry
    }
    // Acknowledge now. Fetching data happens later, outside this request.
    response.writeHead(204).end();
  });

import { fetchAllCyclingActivities, getActivity, withRateLimitRetry } from "./activity-client";
import { getAccessToken, type ConnectionStore, type StoredConnection } from "./token-manager";
import { Walk15Error, type Walk15Config } from "./walk15-config";
import type { Walk15Event } from "./webhook-receiver";

export type Dataset = "steps" | "cycling_daily" | "cycling_activities";

export interface SnapshotStore {
  findConnection(grantId: string): Promise<StoredConnection | undefined>;
  /** Drop tokens and mark the connection ended — only if it is still on this version. */
  endConnection(grantId: string, connectionVersion: string): Promise<void>;
  /** Replace all stored records of a dataset for from..to (inclusive) in one transaction. */
  replace(grantId: string, dataset: Dataset, from: string, to: string, records: unknown[]): Promise<void>;
}

/** "retry_later" keeps the inbox row; expire such rows after a reasonable time (e.g. 24 hours). */
export const processWalk15Event = async (
  config: Walk15Config,
  connections: ConnectionStore,
  snapshots: SnapshotStore,
  event: Walk15Event,
): Promise<"done" | "retry_later"> => {
  const grantId = event.data.grant_id;
  const connection = await snapshots.findConnection(grantId);
  if (!connection) return "retry_later"; // your callback may not have stored the tokens yet

  const eventVersion = BigInt(event.data.connection_version);
  const storedVersion = BigInt(connection.connection_version);
  if (eventVersion < storedVersion) return "done"; // about an older authorization: ignore
  if (eventVersion > storedVersion) return "retry_later"; // newer authorization not exchanged yet

  if (event.type === "connection.revoked") {
    await snapshots.endConnection(grantId, event.data.connection_version);
    return "done";
  }
  if (event.type !== "activity.updated" || !event.data.from || !event.data.to) return "done";

  const from = event.data.from < connection.history_from ? connection.history_from : event.data.from;
  const to = event.data.to;
  const granted = connection.scope.split(" ");
  const scopes = (event.data.scopes ?? []).filter((scope) => granted.includes(scope));
  const token = () => getAccessToken(config, connections, grantId);

  try {
    // Fetch everything first; a failed read must never publish a partial snapshot.
    const staged: [Dataset, unknown[]][] = [];
    for (const scope of scopes) {
      if (scope === "daily_steps:read") {
        const steps = await withRateLimitRetry(async () =>
          getActivity(config, await token(), grantId, "daily-activity", { from, to }),
        );
        staged.push(["steps", steps.data]);
      } else if (scope === "cycling_daily:read") {
        const cycling = await withRateLimitRetry(async () =>
          getActivity(config, await token(), grantId, "cycling-daily", { from, to }),
        );
        staged.push(["cycling_daily", cycling.data]);
      } else if (scope === "cycling_activities:read") {
        staged.push(["cycling_activities", await fetchAllCyclingActivities(config, token, grantId, from, to)]);
      }
    }
    for (const [dataset, records] of staged) await snapshots.replace(grantId, dataset, from, to, records);
    return "done";
  } catch (error) {
    // The user must reconnect before anything can be fetched. Everything else — including
    // invalid_client and a rejected access token — keeps the inbox row and retries with backoff.
    if (error instanceof Walk15Error && error.code === "reauthorization_required") return "done";
    throw error;
  }
};