Custom channel adapter

import { createHmac, randomUUID, timingSafeEqual } from "node:crypto"
import { createServer } from "node:http"

import { Checkfu } from "@checkfu/sdk"

const checkfu = new Checkfu({ apiKey: required("CHECKFU_API_KEY") })
const channelKey = "acme"

// Paired once, out of band. These three ids are the adapter's configuration.
const installationId = required("ACME_EXTERNAL_INSTALLATION_ID")
const placementId = required("ACME_AGENT_INSTALLATION_ID")
const scopeId = required("ACME_SURFACE_SCOPE_ID")
const signingSecret = required("ACME_SIGNING_SECRET")

/**
 * Read one required variable at boot. A missing secret must stop the process,
 * not surface later as a confusing crash inside signature verification on the
 * first real delivery.
 */
function required(name) {
  const value = process.env[name]
  if (value === undefined || value === "") throw new Error(`acme-adapter: ${name} is required`)
  return value
}

/**
 * Yours, always. Checkfu owns your provider's app registration no more than
 * it owns your provider's uptime, so it cannot verify this for you. A forged
 * payload that gets past here is a real message to the platform.
 */
const verified = (raw: Buffer, signature: string): boolean => {
  const expected = createHmac("sha256", signingSecret).update(raw).digest()
  // `Buffer.from` on malformed hex silently truncates rather than throwing, so
  // the length check is load-bearing: without it a short forged signature
  // could reach a comparison it was never long enough to lose.
  const provided = Buffer.from(signature, "hex")
  return expected.length === provided.length && timingSafeEqual(expected, provided)
}

const handleTurn = async (event) => {
  // 1. The speaker becomes a Principal through stable provider ids only.
  const identity = await checkfu.collaboration.externalIdentities.resolve({
    external_installation_id: installationId,
    external_user_id: `${channelKey}:${event.user_id}`,
    display_name: event.user_display_name,
  })

  // 2. One canonical conversation, create-or-return. `kind` is required:
  //    `thread` carries a thread id; `ambient` and `direct` carry null.
  const conversation = {
    kind: "thread",
    root_external_id: `${channelKey}:${event.channel_id}`,
    thread_external_id: event.thread_id ?? event.message_id,
  }
  await checkfu.collaboration.conversationBindings.resolve({
    external_installation_id: installationId,
    agent_installation_id: placementId,
    surface_scope_id: scopeId,
    conversation,
    principal: identity.principal.id,
  })

  // 3. Exactly one authored event, idempotent on the provider's own id. No
  //    client-side retry: a provider redelivery is the retry path, and it
  //    converges server-side on external_event_id.
  const receipt = await checkfu.collaboration.conversationBindings.ingest({
    external_installation_id: installationId,
    agent_installation_id: placementId,
    surface_scope_id: scopeId,
    conversation,
    external_event_id: `${channelKey}:${event.message_id}`,
    principal: identity.principal.id,
    input: { kind: "invoke", content: event.text, is_direct_mention: event.mentions_bot },
  })
  if (receipt.disposition !== "driven") return // filtered or already seen

  // 4. The receipt's coordinates are an EXCLUSIVE cursor. Read strictly after
  //    event_sequence, so a reused conversation never replays an earlier
  //    turn's answer. The response type makes both coordinates nullable
  //    because no-drive dispositions carry null, so a driven receipt missing
  //    either one is a corrupt response: fail loudly rather than letting
  //    `null` become a zero that replays the whole Session.
  if (receipt.event_sequence === null || receipt.event_id === null) {
    throw new Error("acme-adapter: driven ingest returned no authored coordinate")
  }

  const reply = await settledReplyAfter(receipt.session_id, receipt.event_sequence)
  if (reply === null) return // best effort: no reply is a real outcome

  // 5. Claim before posting. The loser of this claim posts nothing.
  const key = {
    conversation_binding_id: receipt.binding.id,
    session_id: receipt.session_id,
    interaction_id: receipt.event_id,
    kind: "final",
    revision: String(reply.throughSequence),
    delivery_role: "reply",
  }
  const claimed = await checkfu.collaboration.conversationPresentations.claim({
    external_installation_id: installationId,
    key,
    through_sequence: reply.throughSequence,
  })
  if (!claimed.won) return

  const dispatchToken = randomUUID()
  const dispatch = {
    external_installation_id: installationId,
    key,
    generation: claimed.attempt.generation,
    owner: claimed.attempt.owner,
    dispatch_token: dispatchToken,
  }
  await checkfu.collaboration.conversationPresentations.beginDispatch(dispatch)

  // 6. The one provider call this adapter is allowed to make for this key.
  //    An outcome you cannot read settles indeterminate — never a blind retry,
  //    because retrying an unknown outcome is how one answer becomes two.
  let outcome
  try {
    await postToAcme(event.channel_id, event.thread_id, reply.text)
    outcome = { state: "posted", provider_message_id: null }
  } catch {
    outcome = { state: "indeterminate" }
  }
  await checkfu.collaboration.conversationPresentations.settle({ ...dispatch, outcome })
}

createServer((request, response) => {
  const chunks = []
  request.on("data", (chunk) => chunks.push(chunk))
  request.on("end", () => {
    const raw = Buffer.concat(chunks)
    if (!verified(raw, request.headers["x-acme-signature"] ?? "")) {
      response.writeHead(401).end()
      return
    }
    // Parse BEFORE acknowledging. A signed body that is not JSON is a real
    // rejection, and parsing it as an argument to handleTurn would throw
    // outside the promise the .catch attaches to — after the 200 was already
    // sent, so the provider would never retry and the process would die.
    let event
    try {
      event = JSON.parse(raw.toString("utf8"))
    } catch {
      response.writeHead(400).end()
      return
    }
    // Acknowledge inside the provider's window; keep the turn supervised.
    response.writeHead(200).end()
    void handleTurn(event).catch((error) => {
      process.stderr.write(`acme-adapter: turn failed: ${String(error)}\n`)
    })
  })
}).listen(8790)