nexus/server/src/services/puter-proxy.ts

import type { Db } from "@paperclipai/db";
import { secretService } from "./secrets.js";
import { costService } from "./costs.js";
import { unprocessable } from "../errors.js";

const PUTER_BASE_URL = "https://api.puter.com/puterai/openai/v1";
const PUTER_DEFAULT_MODEL = "claude-3-5-haiku-20241022";
const PUTER_TOKEN_SECRET_NAME = "puter_auth_token";

export function puterProxyService(db: Db) {
  const secrets = secretService(db);
  const costs = costService(db);

  async function storeToken(companyId: string, token: string) {
    const existing = await secrets.getByName(companyId, PUTER_TOKEN_SECRET_NAME);
    if (existing) {
      return secrets.rotate(existing.id, { value: token });
    }
    return secrets.create(companyId, {
      name: PUTER_TOKEN_SECRET_NAME,
      provider: "local_encrypted",
      value: token,
      description: "Puter.com auth token for AI proxy",
    });
  }

  async function resolveToken(companyId: string): Promise<string> {
    const secret = await secrets.getByName(companyId, PUTER_TOKEN_SECRET_NAME);
    if (!secret) {
      throw unprocessable("Puter auth token not configured");
    }
    return secrets.resolveSecretValue(companyId, secret.id, "latest");
  }

  async function* chatStream(
    companyId: string,
    agentId: string | null | undefined,
    messages: unknown[],
    model: string | undefined,
    signal: AbortSignal | undefined,
  ): AsyncGenerator<string> {
    const token = await resolveToken(companyId);
    const resolvedModel = model ?? PUTER_DEFAULT_MODEL;

    const response = await fetch(`${PUTER_BASE_URL}/chat/completions`, {
      method: "POST",
      headers: {
        Authorization: `Bearer ${token}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        model: resolvedModel,
        messages,
        stream: true,
        stream_options: { include_usage: true },
      }),
      signal,
    });

    if (!response.ok || !response.body) {
      const text = await response.text();
      throw new Error(`Puter API error ${response.status}: ${text}`);
    }

    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let inputTokens = 0;
    let outputTokens = 0;
    let buffer = "";

    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split("\n");
        // Keep last incomplete line in buffer
        buffer = lines.pop() ?? "";

        for (const line of lines) {
          const trimmed = line.trim();
          if (!trimmed.startsWith("data: ")) continue;
          const data = trimmed.slice("data: ".length);
          if (data === "[DONE]") continue;

          let chunk: any;
          try {
            chunk = JSON.parse(data);
          } catch {
            continue;
          }

          if (chunk.usage) {
            inputTokens = chunk.usage.prompt_tokens ?? inputTokens;
            outputTokens = chunk.usage.completion_tokens ?? outputTokens;
          }

          const content = chunk.choices?.[0]?.delta?.content;
          if (content) {
            yield content;
          }
        }
      }
    } finally {
      reader.releaseLock();

      if (agentId) {
        costs
          .createEvent(companyId, {
            agentId,
            provider: "puter",
            biller: "puter",
            billingType: "subscription_included",
            model: resolvedModel,
            inputTokens,
            outputTokens,
            costCents: 0,
            occurredAt: new Date(),
          })
          .catch(() => {});
      }
    }
  }

  return { storeToken, resolveToken, chatStream };
}