transport rpc

datasource db {
  provider = "postgresql"
  url = env("DATABASE_URL")
}

auth Operator {
  id Int
}

model Widget {
  id Int @id
  name String

  @@allow("read", auth() != null)
  @@allow("create", auth() != null)
  @@allow("update", auth() != null)
  @@allow("delete", auth() != null)
}

procedure ping(args: PingArgs): PingArgs
  @allow(auth() != null)

use cratestack::axum::Router;

let app: Router = cratestack_schema::axum::rpc_router(
    db,
    MyProcedures,
    CodecSet::new(CborCodec, JsonCodec),
    MyAuthProvider,
);

// POST /rpc/model.Widget.create
// body = CreateWidgetInput directly (same as the REST POST body shape)
{ "name": "left handle" }

// POST /rpc/model.Widget.get  or  /rpc/model.Widget.delete
{ "id": 42 }

// get (not delete) also accepts the REST fetch-query surface, all optional —
// same keys and semantics as GET /<plural>/{id}; old bare {id} frames
// keep working unchanged
{
  "id": 42,
  "fields": ["id", "name"],
  "include": ["parts"],
  "include_fields": { "parts": ["id"] },
  "computedParams": "{\"proxyUrl\":{\"width\":800}}"
}

// POST /rpc/model.Widget.update
// patch decoded against the model's UpdateWidgetInput shape
{ "id": 42, "patch": { "name": "new name" } }

// POST /rpc/model.Widget.list
// mirrors the REST URL query 1:1 — same keys, same semantics
{
  "limit": 20,
  "offset": 40,
  "fields": ["id", "name"],
  "include": ["..."],
  "sort": "name desc",
  "where": "...",
  "filters": [{ "key": "active", "value": "true" }]
}

// POST /rpc/procedure.ping
// body = procedure Args directly
{ "args": { "nonce": "abc" } }

// request body — a sequence of RpcRequest frames
[
  { "id": 1, "op": "procedure.ping",       "input": { "args": { "nonce": "a" } } },
  { "id": 2, "op": "model.Widget.create",  "input": { "name": "frame two" }, "idem": "client-key-7b3" },
  { "id": 3, "op": "model.Widget.get",     "input": { "id": 42 } }
]

// response body — a sequence of RpcResponseFrame, same order as the request
[
  { "id": 1, "output": { "nonce": "a" } },
  { "id": 2, "output": { "id": 17, "name": "frame two" } },
  { "id": 3, "error":  { "code": "not_found", "message": "widget 42" } }
]

POST /rpc/procedure.manyPings HTTP/1.1
Content-Type: application/cbor
Accept: application/cbor-seq

procedure ticks(args: TickerArgs): Tick[]
  @allow(auth() != null)
  @stream

use cratestack::client_rust::rpc::{RpcClient, RpcClientError};

let mut rx = rpc_client
    .call_streaming::<TicksArgs, Tick>("procedure.ticks", &TicksArgs { count: 100 })
    .await?;

while let Some(item) = rx.recv().await {
    match item {
        Ok(tick) => render(tick),
        Err(RpcClientError::Remote(err)) => {
            // Per-item error — terminal. The next recv() will return None.
            eprintln!("server returned {}: {}", err.body.code, err.body.message);
            break;
        }
        Err(other) => {
            eprintln!("transport/decode error: {other}");
            break;
        }
    }
}

use cratestack_client_flutter::{FlutterChunkWire, FlutterHeader, FlutterRuntime, FlutterRuntimeError};
use flutter_rust_bridge::frb;

#[frb(sync)]
pub fn rpc_call_streamed(
    runtime: &FlutterRuntime,
    op_id: String,
    input: Vec<u8>,
    headers: Vec<FlutterHeader>,
    sink: flutter_rust_bridge::StreamSink<FlutterChunkWire>,
) -> Result<(), FlutterRuntimeError> {
    runtime.rpc_call_streamed(&op_id, input, headers, move |chunk| sink.add(chunk).is_ok())
}

await for (final chunk in stream) {
  switch (chunk) {
    case FlutterChunkWire_Item(:final field0):
      final tick = Tick.fromWire(cbor.cbor.decode(field0));
      renderRow(tick);
    case FlutterChunkWire_End():
      break;
    case FlutterChunkWire_Error(:final field0):
      handleError(field0);
      break;
  }
}

final decoder = FlutterCborSeqDecoder();
final response = await dio.post<ResponseBody>(
  '/rpc/$opId',
  data: encodedInput,
  options: Options(
    responseType: ResponseType.stream,
    contentType: 'application/cbor',
    headers: {'Accept': 'application/cbor-seq'},
  ),
);

final items = response.data!.stream
    .transform(CborSeqStreamTransformer(decoder))
    .map((bytes) => Tick.fromWire(cbor.cbor.decode(bytes)));

await for (final tick in items) renderRow(tick);

final dio = Dio(BaseOptions(baseUrl: baseUrl))
  ..interceptors.add(InterceptorsWrapper(onRequest: (opts, h) {
    opts.headers['Authorization'] = 'Bearer ${currentToken()}';
    opts.headers.putIfAbsent('Idempotency-Key', () => const Uuid().v4());
    h.next(opts);
  }))
  ..interceptors.add(RetryInterceptor(dio: dio, retries: 3)); // dio_smart_retry

import { ExampleWidgetClientClient, CratestackRpcStreamError } from "@example/widget-client";

const client = new ExampleWidgetClientClient("https://api.example.com", { basePath: "/api" });

try {
  for await (const tick of client.runtime.stream<Tick>("procedure.ticks", { count: 100 })) {
    renderRow(tick);
  }
} catch (error) {
  if (error instanceof CratestackRpcStreamError) {
    // Mid-stream sentinel (tag 48900) — the stream ends here either way.
    handleError(error.body);
  } else {
    // Transport failure: network error, malformed/truncated cbor-seq body.
    throw error;
  }
}

// createLoggerStreamLink is generated alongside the runtime (src/links.ts)
// and re-exported from the package root, same as the runtime class itself.
import { ExampleWidgetClientClient, createLoggerStreamLink } from "@example/widget-client";

const client = new ExampleWidgetClientClient("https://api.example.com", {
  streamLinks: [createLoggerStreamLink()],
});

{
  "code": "not_found",
  "message": "widget 42",
  "details": null
}

export interface RpcLinkRequest {
  readonly kind: "unary" | "batch";
  readonly opId: string;
  readonly input: unknown;           // raw, pre-codec-encode
  readonly headers: Headers;
  readonly signal: AbortSignal | null;
  readonly idempotencyKey?: string;
  readonly codec: CratestackRpcCodec;
  readonly fetchFn: typeof fetch;
  readonly urls: { unary(opId: string): string; batch(): string };
}

export interface RpcLinkResponse { readonly response: Response; }

export type RpcLinkNext = (request: RpcLinkRequest) => Promise<RpcLinkResponse>;

export type RpcLink = (request: RpcLinkRequest, next: RpcLinkNext) => Promise<RpcLinkResponse>;

// Reference link, ships in every generated project:
export function createLoggerLink(logger?: Pick<Console, "info" | "error">): RpcLink;

import { ExampleWidgetClientClient } from "@example/widget-client";
import { createLoggerLink } from "@example/widget-client";

const client = new ExampleWidgetClientClient("https://api.example.com", {
  links: [createLoggerLink()],
});

import { createBatchLink, createLoggerLink } from "@cratestack/api";
import { CratestackRpcRuntime } from "./generated/runtime";

const runtime = new CratestackRpcRuntime("https://api.example.com", {
  links: [createLoggerLink(), createBatchLink()],
});
const client = new MyGeneratedClient(runtime);

// These three calls, issued in the same tick, become ONE /rpc/batch request:
const [a, b, c] = await Promise.all([
  client.widgets.get(1), client.widgets.get(2), client.widgets.get(3),
]);