Connect a user

import { realpathSync } from "node:fs";
import { pathToFileURL } from "node:url";

// Exchange a callback code. Call it once per code; never retry automatically.
export async function exchangeCode({ code, verifier }) {
  const response = await fetch(`${process.env.CORRCONNECT_BASE}/oauth/token`, {
    method: "POST",
    body: new URLSearchParams({
      grant_type: "authorization_code",
      client_id: process.env.CLIENT_ID,
      code,
      redirect_uri: process.env.REDIRECT_URI,
      code_verifier: verifier,
    }),
  });
  const body = await response.json();
  if (!response.ok) {
    throw new Error(`token exchange failed: ${body.error} (${body.request_id})`);
  }
  const profile = await fetch(`${process.env.CORRCONNECT_BASE}/v1/profile`, {
    headers: { Authorization: `Bearer ${body.access_token}` },
  }).then((r) => r.json());
  return {
    sub: profile.sub,
    accessToken: body.access_token,
    accessExpiresAt: Date.now() + body.expires_in * 1000,
    refreshToken: body.refresh_token ?? null,
    scopes: body.scope.split(" "),
  };
}

if (import.meta.url === pathToFileURL(realpathSync(process.argv[1])).href) {
  console.log(await exchangeCode({ code: process.env.CODE, verifier: process.env.CODE_VERIFIER }));
}

import os
import time

import requests

BASE = os.environ["CORRCONNECT_BASE"]


def exchange_code(code: str, verifier: str) -> dict:
    """Exchange a callback code. Call it once per code; never retry automatically."""
    response = requests.post(f"{BASE}/oauth/token", data={
        "grant_type": "authorization_code",
        "client_id": os.environ["CLIENT_ID"],
        "code": code,
        "redirect_uri": os.environ["REDIRECT_URI"],
        "code_verifier": verifier,
    }, timeout=10)
    body = response.json()
    if response.status_code != 200:
        raise RuntimeError(f"token exchange failed: {body['error']} ({body['request_id']})")
    profile = requests.get(f"{BASE}/v1/profile", timeout=10,
                           headers={"Authorization": f"Bearer {body['access_token']}"}).json()
    return {
        "sub": profile["sub"],
        "access_token": body["access_token"],
        "access_expires_at": time.time() + body["expires_in"],
        "refresh_token": body.get("refresh_token"),
        "scopes": body["scope"].split(),
    }


if __name__ == "__main__":
    print(exchange_code(os.environ["CODE"], os.environ["CODE_VERIFIER"]))

import { realpathSync } from "node:fs";
import { pathToFileURL } from "node:url";

const inFlight = new Map(); // sub -> Promise, so one refresh per athlete at a time

export function refreshFor(sub, refreshToken) {
  if (!inFlight.has(sub)) {
    const attempt = (async () => {
      const response = await fetch(`${process.env.CORRCONNECT_BASE}/oauth/token`, {
        method: "POST",
        body: new URLSearchParams({
          grant_type: "refresh_token",
          client_id: process.env.CLIENT_ID,
          refresh_token: refreshToken,
        }),
      });
      const body = await response.json();
      if (!response.ok) {
        // invalid_grant: the grant ended or the token was spent. Ask the athlete to connect again.
        throw Object.assign(new Error(body.error), { oauthError: body.error });
      }
      return body; // save body.refresh_token before using body.access_token
    })().finally(() => inFlight.delete(sub));
    inFlight.set(sub, attempt);
  }
  return inFlight.get(sub);
}

if (import.meta.url === pathToFileURL(realpathSync(process.argv[1])).href) {
  console.log(await refreshFor("example-sub", process.env.REFRESH_TOKEN));
}

import os
import threading
from collections import defaultdict

import requests

_locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)


def refresh_for(sub: str, load_refresh_token, save_tokens) -> dict:
    """Refresh one athlete's tokens; concurrent callers for the same sub wait their turn."""
    with _locks[sub]:
        response = requests.post(os.environ["CORRCONNECT_BASE"] + "/oauth/token", data={
            "grant_type": "refresh_token",
            "client_id": os.environ["CLIENT_ID"],
            "refresh_token": load_refresh_token(sub),  # read inside the lock: a waiter gets the new one
        }, timeout=10)
        body = response.json()
        if response.status_code != 200:
            raise PermissionError(body["error"])  # invalid_grant: ask the athlete to connect again
        save_tokens(sub, body)  # store the new refresh token before using the access token
        return body


if __name__ == "__main__":
    store = {"example-sub": os.environ["REFRESH_TOKEN"]}
    print(refresh_for("example-sub", store.get, lambda sub, body: store.update({sub: body["refresh_token"]})))

import Foundation

/// Keeps refresh tokens and runs at most one refresh per athlete at a time.
actor TokenStore {
    struct Tokens: Decodable {
        let access_token: String
        let expires_in: Int
        let scope: String
        let refresh_token: String?
        let refresh_token_expires_in: Int?
    }

    private var refreshTokens: [String: String] = [:]
    private var inFlight: [String: Task<Tokens, Error>] = [:]
    private let base: URL
    private let clientID: String

    init(base: URL, clientID: String) {
        self.base = base
        self.clientID = clientID
    }

    func save(sub: String, refreshToken: String) {
        refreshTokens[sub] = refreshToken
    }

    func refresh(sub: String) async throws -> Tokens {
        if let running = inFlight[sub] { return try await running.value }
        guard let current = refreshTokens[sub] else { throw URLError(.userAuthenticationRequired) }
        let task = Task { try await Self.post(base: base, clientID: clientID, refreshToken: current) }
        inFlight[sub] = task
        defer { inFlight[sub] = nil }
        do {
            let tokens = try await task.value
            refreshTokens[sub] = tokens.refresh_token  // save before using the access token
            return tokens
        } catch {
            refreshTokens[sub] = nil  // invalid_grant: the athlete has to connect again
            throw error
        }
    }

    private static func post(base: URL, clientID: String, refreshToken: String) async throws -> Tokens {
        var form = URLComponents()
        form.queryItems = [
            URLQueryItem(name: "grant_type", value: "refresh_token"),
            URLQueryItem(name: "client_id", value: clientID),
            URLQueryItem(name: "refresh_token", value: refreshToken),
        ]
        var request = URLRequest(url: base.appendingPathComponent("oauth/token"))
        request.httpMethod = "POST"
        request.setValue("application/x-www-form-urlencoded", forHTTPHeaderField: "Content-Type")
        request.httpBody = form.percentEncodedQuery!.replacingOccurrences(of: "+", with: "%2B").data(using: .utf8)
        let (data, response) = try await URLSession.shared.data(for: request)
        guard (response as? HTTPURLResponse)?.statusCode == 200 else {
            throw URLError(.userAuthenticationRequired)
        }
        return try JSONDecoder().decode(Tokens.self, from: data)
    }
}

let env = ProcessInfo.processInfo.environment
let store = TokenStore(base: URL(string: env["CORRCONNECT_BASE"]!)!, clientID: env["CLIENT_ID"]!)
await store.save(sub: "example-sub", refreshToken: env["REFRESH_TOKEN"]!)
async let first = store.refresh(sub: "example-sub")
async let second = store.refresh(sub: "example-sub")  // joins the first request
let (a, b) = try await (first, second)
print(a.access_token == b.access_token, a.scope)