Partners

import { createHmac, createHash, randomUUID } from "node:crypto"

const PARTNER_ID = process.env.LATITUDE_PARTNER_ID
const PARTNER_SECRET = process.env.LATITUDE_PARTNER_SECRET

export async function provisionAccount({ user, organization }) {
  const pathname = `/v1/private/partners/${PARTNER_ID}/accounts`
  // Serialize once: the signature covers these exact bytes.
  const body = JSON.stringify({ user, organization })

  const timestamp = String(Math.floor(Date.now() / 1000))
  const nonce = randomUUID()
  const bodyHash = createHash("sha256").update(body, "utf8").digest("hex")
  const stringToSign = `v1:${timestamp}:POST:${pathname}:${nonce}:${bodyHash}`
  const signature = createHmac("sha256", PARTNER_SECRET).update(stringToSign, "utf8").digest("hex")

  const response = await fetch(`https://api.latitude.so${pathname}`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-Partner-Timestamp": timestamp,
      "X-Partner-Signature": `v1=${signature}`,
      "X-Partner-Nonce": nonce,
    },
    body,
  })

  if (response.status === 409) throw new Error("account_already_exists")
  if (!response.ok) throw new Error(`Provisioning failed: ${response.status}`)

  return response.json()
}

import hashlib
import hmac
import json
import os
import time
import uuid

import requests

PARTNER_ID = os.environ["LATITUDE_PARTNER_ID"]
PARTNER_SECRET = os.environ["LATITUDE_PARTNER_SECRET"]


def provision_account(user: dict, organization: dict | None = None) -> dict:
    pathname = f"/v1/private/partners/{PARTNER_ID}/accounts"
    payload: dict = {"user": user}
    if organization:
        payload["organization"] = organization

    # Serialize once: the signature covers these exact bytes.
    body = json.dumps(payload)

    timestamp = str(int(time.time()))
    nonce = str(uuid.uuid4())
    body_hash = hashlib.sha256(body.encode()).hexdigest()
    string_to_sign = f"v1:{timestamp}:POST:{pathname}:{nonce}:{body_hash}"
    signature = hmac.new(PARTNER_SECRET.encode(), string_to_sign.encode(), hashlib.sha256).hexdigest()

    response = requests.post(
        f"https://api.latitude.so{pathname}",
        data=body,
        headers={
            "Content-Type": "application/json",
            "X-Partner-Timestamp": timestamp,
            "X-Partner-Signature": f"v1={signature}",
            "X-Partner-Nonce": nonce,
        },
    )

    if response.status_code == 409:
        raise RuntimeError("account_already_exists")
    response.raise_for_status()
    return response.json()