Functions

// functions/createMessage.ts
import { mutation, v } from "@pylonsync/functions";

export default mutation({
  // Every function declares an auth mode. "user" is the default;
  // see the Auth section below. The framework enforces this before
  // the handler runs, so `ctx.auth.userId` is `string` here.
  args: {
    roomId: v.id("Room"),
    body: v.string(),
  },
  async handler(ctx, args) {
    const id = await ctx.db.insert("Message", {
      roomId: args.roomId,
      authorId: ctx.auth.userId,
      body: args.body,
      sentAt: new Date().toISOString(),
    });
    return { id };
  },
});

ctx.auth.userId   // string (when auth: "user" | "admin") OR string | null (when auth: "public" | "guest")
ctx.auth.isAdmin  // boolean
ctx.auth.tenantId // string | null  (selected org)
ctx.auth.roles    // string[] (exact active-org/session roles)
ctx.auth.isGuest  // boolean — true for an anonymous guest session
ctx.auth.elevate({ admin: true, reason: "..." })  // promote AFTER verifying a webhook/HMAC
// No ctx.auth.email on the handler ctx — fetch the user row for email:
//   const u = await ctx.db.get("User", ctx.auth.userId)

// ctx.db — on queries (read-only) and mutations (read + write). NOT on actions.
ctx.db.insert(entity, data)
ctx.db.get(entity, id)
ctx.db.query(entity, filter?)
ctx.db.list(entity)                   // whole entity (no filter)
ctx.db.update(entity, id, patch)
ctx.db.delete(entity, id)
ctx.db.link(entity, id, relation, targetId)
ctx.db.unlink(entity, id, relation)

// Mutations + actions only (not queries):
ctx.error(code, message)              // throw typed error → mutation rolls back
ctx.scheduler.runAfter(delayMs, ...)  // schedule (deferred until COMMIT)
ctx.llm.complete(request)             // LLM completion, returns when the model finishes
ctx.llm.stream(request, onEvent)      // LLM completion, event per delta
ctx.stream.write(text)                // SSE chunk → the one client holding this request
ctx.rooms.broadcast(room, topic, data) // push → every subscriber of a presence room

// All three flavors:
ctx.files.signedUrl(fileId, { ttlSecs: 300 })  // mint a short-lived signed download URL
ctx.files.info(fileId)                         // content type, owner, public, or null
ctx.shards.ticket(shardId, { subscriberId, claims, ttlSecs })  // mint a signed realtime shard ticket
ctx.shards.get(shardId) / ctx.shards.list()                    // running shards

// Actions only:
ctx.shards.create(kind, shardId, params)  // start a WebAssembly shard (see Realtime shards)
ctx.shards.stop(shardId)                  // stop it and close its connections
ctx.files.store(data, { name, contentType, public })  // write a file to the app's storage
ctx.files.delete(fileId)                  // delete a stored file
ctx.files.read(fileId)                    // a stored file's bytes, with server trust
ctx.crdt.snapshot/since/apply(entity, id, ...)  // edit a row's live CRDT document (see Loro CRDTs)
ctx.agents.run(name, { input, as, context })  // run an agent() as a named user (see Agents)

export default query({
  args: { eventId: v.id("Event"), fileId: v.string() },
  async handler(ctx, args) {
    const event = await ctx.db.get("Event", args.eventId);
    await ctx.requireMember(event.orgId, { role: "organizer" });
    return { url: await ctx.files.signedUrl(args.fileId, { ttlSecs: 300 }) };
  },
});

export default action({
  args: { raceId: v.id("Race"), jpegBase64: v.string() },
  async handler(ctx, args) {
    const photo = await ctx.files.store(Buffer.from(args.jpegBase64, "base64"), {
      name: "finish.jpg",
      contentType: "image/jpeg",
    });
    await ctx.runMutation("setRacePhoto", { raceId: args.raceId, fileId: photo.id });
    return photo;
  },
});

// Send an uploaded PDF to the model.
const file = await ctx.files.read(attachment.fileId);
await ctx.llm.complete({
  messages: [{ role: "user", content: [
    { type: "text", text: "Summarize this." },
    { type: "document", title: attachment.name, source: { type: "base64", media_type: "application/pdf", data: Buffer.from(file.data).toString("base64") } },
  ] }],
});

export default mutation({
  args: { characterId: v.id("Character") },
  async handler(ctx, args) {
    const c = await ctx.db.get("Character", args.characterId);
    if (c?.ownerId !== ctx.auth.userId) throw ctx.error("FORBIDDEN", "not your character");
    const shard = `zone-${c.zone}`;
    const ticket = await ctx.shards.ticket(shard, {
      subscriberId: c.id,
      claims: { character: c.id, realm: c.realm },
    });
    return { shard, ticket };
  },
});

export default mutation({
  auth: "user",   // default — signed-in user required
  args: { ... },
  async handler(ctx, args) {
    // ctx.auth.userId is `string`, not `string | null`. The
    // redundant null check is gone.
    await ctx.db.insert("Note", { authorId: ctx.auth.userId, ... });
  },
});

// Public endpoint — has to be explicit, never the default.
export default action({
  auth: "public",
  async handler(ctx) {
    // ctx.auth.userId is `string | null`. Check it if you need it.
    return { ok: true };
  },
});

PYLON_STRICT_FN_POLICIES=1 pylon dev

// functions/stripeWebhook.ts
// The webhook is an action — only actions get ctx.request (the raw
// body a signature check needs). Actions have no ctx.db, so the DB
// work runs in an internal mutation the action hands off to.
export default action({
  auth: "public",  // webhook — provider doesn't sign in
  async handler(ctx) {
    const sig = ctx.request!.headers["stripe-signature"];
    const ok = verifyStripeSignature(secret, ctx.request!.rawBody, sig);
    if (!ok) throw ctx.error("INVALID_SIGNATURE", "bad sig");
    // Trust boundary crossed — the webhook is the trusted caller.
    await ctx.runMutation("settleStripeEvent", { customerId });
  },
});

// functions/settleStripeEvent.ts — internal: unreachable over HTTP
export default mutation({
  internal: true,
  args: { customerId: v.string() },
  async handler(ctx, args) {
    // Use ctx.db.unsafe to read any org's row regardless of who's
    // calling. Comment justifies the bypass for code review.
    const org = await ctx.db.unsafe.lookup("Org", "stripeCustomerId", args.customerId);
    // ...
  },
});

v.string()              v.int()                v.float()
v.boolean()             v.datetime()           v.id("User")
v.optional(v.string())  v.array(v.string())    v.literal("open" | "closed")
v.object({ k: v.string() })

// functions/listIssues.ts
import { query, v } from "@pylonsync/functions";

export default query({
  args: { teamId: v.id("Team") },
  async handler(ctx, args) {
    return ctx.db.query("Issue", { teamId: args.teamId });
  },
});

// functions/sendEmail.ts
import { action, v } from "@pylonsync/functions";

export default action({
  args: { to: v.string(), subject: v.string() },
  async handler(ctx, args) {
    await fetch("https://api.resend.com/emails", {
      method: "POST",
      headers: { Authorization: `Bearer ${process.env.RESEND_KEY}` },
      body: JSON.stringify({ to: args.to, subject: args.subject }),
    });
    return { ok: true };
  },
});

// functions/twilioSms.ts
import { action } from "@pylonsync/functions";
import { createHmac, timingSafeEqual } from "node:crypto";

export default action({
  auth: "public",
  async handler(ctx) {
    const req = ctx.request!;
    const params = new URLSearchParams(req.rawBody);

    // Twilio signs the full URL plus the POST params sorted by name.
    const signed =
      req.url +
      [...params.keys()].sort().map((k) => k + params.get(k)).join("");
    const expected = createHmac("sha1", process.env.TWILIO_AUTH_TOKEN!)
      .update(signed)
      .digest("base64");
    const given = req.headers["x-twilio-signature"] ?? "";
    if (
      given.length !== expected.length ||
      !timingSafeEqual(Buffer.from(given), Buffer.from(expected))
    ) {
      return ctx.response({ status: 403, body: "bad signature" });
    }

    const leadId = await ctx.runMutation("recordInboundSms", {
      from: params.get("From"),
      body: params.get("Body"),
    });
    const runs = await ctx.workflows.list({ key: leadId, status: "active" });
    for (const run of runs) {
      await ctx.workflows.sendEvent(run.id, "seller_replied", {
        body: params.get("Body"),
      });
    }
    return ctx.response({ contentType: "text/xml", body: "<Response/>" });
  },
});

// functions/ask.ts
import { action, v } from "@pylonsync/functions";

export default action({
  args: { question: v.string() },
  async handler(ctx, args) {
    const res = await ctx.llm.stream(
      { messages: [{ role: "user", content: args.question }] },
      (e) => {
        // Push each token to the caller as the model produces it.
        if (e.type === "text_delta") ctx.stream.write(e.text);
      },
    );
    return { stopReason: res.stop_reason, usage: res.usage };
  },
});

for await (const chunk of db.streamFn("ask", { question: "why is the sky blue?" })) {
  setAnswer((prev) => prev + chunk);
}

const res = await ctx.llm.stream({ messages }, onEvent, { signal });

export default action({
  timeout: 300,   // idle seconds; total run capped at 3000 s
  args: { ... },
  async handler(ctx, args) { ... },
});

// functions/agent.ts
import { action, v } from "@pylonsync/functions";

// Anthropic Messages shape: content is prose, or content blocks —
// tool_use coming back from the model, tool_result going the other way.
type Turn = { role: "user" | "assistant"; content: string | any[] };

const tools = [
  {
    name: "get_order",
    description: "Look up an order by id.",
    input_schema: {
      type: "object",
      properties: { orderId: { type: "string" } },
      required: ["orderId"],
    },
  },
];

export default action({
  args: { question: v.string() },
  timeout: 300,   // a slow tool call can be silent for minutes
  async handler(ctx, args) {
    const messages: Turn[] = [{ role: "user", content: args.question }];

    for (let turn = 0; turn < 10; turn++) {
      const res = await ctx.llm.stream({ messages, tools }, (e) => {
        if (e.type === "text_delta") ctx.stream.write(e.text);
      });

      // Keep the assistant turn verbatim — it carries the tool_use
      // blocks the next turn's tool_result blocks refer to by id.
      messages.push({ role: "assistant", content: res.content });

      if (res.stop_reason !== "tool_use") return { turns: turn + 1 };

      const results = [];
      for (const block of res.content) {
        if (block.type !== "tool_use") continue;
        const order = await ctx.runQuery("getOrder", {
          id: (block.input as { orderId: string }).orderId,
        });
        results.push({
          type: "tool_result" as const,
          tool_use_id: block.id,
          content: JSON.stringify(order ?? { error: "not found" }),
        });
      }
      messages.push({ role: "user", content: results });
    }

    throw ctx.error("AGENT_LOOP_LIMIT", "tool loop did not converge");
  },
});

await ctx.rooms.broadcast("chat-42", "agent.delta", { text: "hi" });
// => { delivered: true }

await ctx.llm.stream({ messages }, (e) => {
  if (e.type !== "text_delta") return;
  ctx.stream.write(e.text);                                  // the caller
  void ctx.rooms.broadcast(room, "agent.delta", { text: e.text }); // everyone else
});

import { useRoom, getSync } from "@pylonsync/react";

useRoom("chat-42", userId);   // joins the room — required to receive pushes

useEffect(() => {
  return getSync().subscribeRoomMessages("chat-42", (m) => {
    if (m.topic === "agent.delta") append((m.payload as { text: string }).text);
  });
}, []);

import { callFn } from "@pylonsync/react";

async function onSend() {
  const { id } = await callFn("createMessage", {
    roomId: "01H...",
    body: "hello",
  });
  console.log("created", id);
}

if (!ctx.auth.userId) {
  throw ctx.error("UNAUTHENTICATED", "log in first");
}
if (msg.length > 2000) {
  throw ctx.error("INVALID_ARGS", "message too long");
}