ts-draft-client.ts

// Original draft: https://gist.github.com/IAmJSD/a8872306d4c3b50369636e17b29330db
import * as v from "valibot";
import * as Y from "yjs";
import { encode, decode } from "@msgpack/msgpack";

import {
  object,
  url,
  string,
  pipe,
  parse,
  number,
  InferOutput,
  array,
  union,
  literal,
  boolean,
  strictTuple,
  optional,
} from "valibot";

export const AuthenticationAndCodeExchangeURL = object({
  authentication_url: pipe(string(), url()),
  code_exchange_url: pipe(string(), url()),
  browser_authentication: optional(object({
    method: literal("post_message"),
    code_challenge_method: literal("S256-hex"),
    origins: array(string()),
  })),
});

/**
 * Returns a authentication and code exchange URL. The URLs MUST not be relative. The "state" parameter is injected into authentication_url.
 * The authentication URL should do the flow in a browser and return with schist://ig-callback?state=<state here>&code=<your code>
 */
export async function getAuthenticationAndCodeExchangeURLs(
  state: string,
  domain: string,
) {
  domain = domain.toLowerCase();
  if (domain.startsWith("http://")) throw new Error("Domain cannot be http");
  if (!domain.startsWith("https://")) domain = `https://${domain}`;

  const schistUrlsUrl = new URL("/.schist/auth-urls.json", domain);
  const res = await fetch(schistUrlsUrl);
  if (!res.ok)
    throw new Error(`/.schist/auth-urls.json returned status ${res.status}`);
  const j = parse(AuthenticationAndCodeExchangeURL, await res.json());

  const authUrl = new URL(j.authentication_url);
  authUrl.searchParams.set("state", state);

  return {
    authentication_url: authUrl.toString(),
    code_exchange_url: j.code_exchange_url,
    browser_authentication: j.browser_authentication,
  };
}

export const CodeExchangeResponse = object({
  access_token: string(),
  refresh_token: string(),
  expires_at: number(),
  generation_endpoint_url: string(),

  /** Optional extension: one authenticated WSS connection for the remote workspace. */
  workspace_websocket_url: optional(pipe(string(), url())),

  /** If the app wishes to logout, it just performs a DELETE to this URL with no additional body/headers/params. */
  logout_url: string(),
});

export const SCHIST_BROWSER_ORIGIN = "https://try.schist.app";
export const SCHIST_CLOUD_ORIGIN = "https://cloud.schist.app";

function officialCloudUrl(raw: string, websocket = false): string {
  const u = new URL(raw);
  if (u.origin !== (websocket ? "wss://cloud.schist.app" : SCHIST_CLOUD_ORIGIN) || u.username || u.password)
    throw new Error("Browser cloud connections are restricted to cloud.schist.app");
  return u.toString();
}
// Only file upload bodies may go to this private Neon storage endpoint.
// Access/refresh tokens, auth callbacks and WebSockets stay on the cloud origin.
export const SCHIST_UPLOAD_ORIGIN = "https://br-bitter-cake-ayadqsoa.storage.c-5.us-east-2.aws.neon.tech";
function officialUploadUrl(raw: string): string {
  const u = new URL(raw);
  if (![SCHIST_CLOUD_ORIGIN, SCHIST_UPLOAD_ORIGIN].includes(u.origin) || u.username || u.password)
    throw new Error("Unrecognized Schist upload endpoint");
  return u.toString();
}
function officialBrowser() {
  if (typeof window === "undefined" || window.location.origin !== SCHIST_BROWSER_ORIGIN)
    throw new Error("Browser cloud is available at try.schist.app");
}
function browserCredentials(credentials: InferOutput<typeof CodeExchangeResponse>) {
  officialCloudUrl(credentials.generation_endpoint_url);
  officialCloudUrl(credentials.logout_url);
  if (!credentials.workspace_websocket_url) throw new Error("No workspace endpoint");
  officialCloudUrl(credentials.workspace_websocket_url, true);
  return credentials;
}

/** Call directly from a click handler; credentials remain in the current tab's memory. */
export async function signIntoSchistCloud() {
  officialBrowser();
  const popup = window.open("about:blank", "_blank", "popup,width=520,height=720");
  if (!popup) throw new Error("Allow the sign-in popup and try again");
  const hex = (bytes: Uint8Array) => Array.from(bytes, b => b.toString(16).padStart(2, "0")).join("");
  const state = crypto.randomUUID();
  const verifier = hex(crypto.getRandomValues(new Uint8Array(32)));
  let listener: ((event: MessageEvent) => void) | undefined;
  let timer: ReturnType<typeof setInterval> | undefined;
  try {
    const response = await fetch(`${SCHIST_CLOUD_ORIGIN}/.schist/auth-urls.json`, {
      credentials: "omit", redirect: "error", signal: AbortSignal.timeout(60_000),
    });
    if (!response.ok) throw new Error(`Cloud discovery failed (${response.status})`);
    const discovery = parse(AuthenticationAndCodeExchangeURL, await response.json());
    if (!discovery.browser_authentication?.origins.includes(SCHIST_BROWSER_ORIGIN))
      throw new Error("Provider does not support this browser sign-in flow");
    const codeExchangeUrl = officialCloudUrl(discovery.code_exchange_url);
    const authorization = new URL(officialCloudUrl(discovery.authentication_url));
    const challenge = hex(new Uint8Array(await crypto.subtle.digest("SHA-256", new TextEncoder().encode(verifier))));
    authorization.searchParams.set("state", state);
    authorization.searchParams.set("return_origin", SCHIST_BROWSER_ORIGIN);
    authorization.searchParams.set("code_challenge", challenge);
    const code = await new Promise<string>((resolve, reject) => {
      const deadline = Date.now() + 600_000;
      listener = event => {
        if (event.origin !== SCHIST_CLOUD_ORIGIN || event.source !== popup) return;
        const data = event.data;
        if (data?.type === "schist.authorization" && data.state === state && typeof data.code === "string" && data.code.length > 0 && data.code.length <= 256)
          resolve(data.code);
      };
      window.addEventListener("message", listener);
      timer = setInterval(() => {
        if (popup.closed || Date.now() >= deadline) reject(new Error("Sign-in closed or expired"));
      }, 100);
      popup.location.href = authorization.toString();
    });
    const credentials = browserCredentials(await performCodeExchange(codeExchangeUrl, code, state, verifier));
    return { codeExchangeUrl, credentials, browserCloud: true as const };
  } finally {
    if (listener) window.removeEventListener("message", listener);
    if (timer) clearInterval(timer);
    popup.close();
  }
}

/**
 * Performs the code exchange with the code_exchange_url we got earlier and the code/state we were
 * just redirected with (called in response to schist://ig-callback?state=<state here>&code=<your code>).
 */
export async function performCodeExchange(
  codeExchangeUrl: string,
  code: string,
  state: string,
  codeVerifier?: string,
) {
  const body = {
    response_type: "code",
    ...(codeVerifier === undefined ? {} : { code_verifier: codeVerifier }),
    code,
    state,
    schist_spec_version: 1,
  };
  const res = await fetch(codeExchangeUrl, {
    credentials: "omit",
    redirect: "error",
    headers: {
      "Content-Type": "application/json",
    },
    body: JSON.stringify(body),
    method: "POST",
  });
  if (!res.ok) throw new Error("failed to exchange code");
  return parse(CodeExchangeResponse, await res.json());
}

/** Perform a refresh with the refresh token and the code_exchange_url we got earlier. */
export async function performTokenRefresh(
  codeExchangeUrl: string,
  refreshToken: string,
) {
  const body = {
    response_type: "refresh_token",
    refresh_token: refreshToken,
    schist_spec_version: 1,
  };
  const res = await fetch(codeExchangeUrl, {
    credentials: "omit",
    redirect: "error",
    headers: {
      "Content-Type": "application/json",
    },
    body: JSON.stringify(body),
    method: "POST",
  });
  if (!res.ok) throw new Error("failed to exchange refresh token");
  return parse(CodeExchangeResponse, await res.json());
}

async function refreshIfNeeded(
  codeExchangeUrl: string,
  body: InferOutput<typeof CodeExchangeResponse>,
) {
  if (Date.now() / 1000 <= body.expires_at) {
    // This is fine
    return null;
  }
  return await performTokenRefresh(codeExchangeUrl, body.refresh_token);
}

/** A single-line/free-form text input the user fills in. The value is sent back keyed by `id`. */
const TextBox = object({
  t: literal("text"),
  title: string(),
  description: string(),

  /** If true, the app should refuse to generate until the user has given a value. */
  required: boolean(),

  /** The key this field's value is sent under in the generation/preview body. */
  id: string(),
});

/** A pick-from-a-list input. The chosen value id(s) are sent back keyed by `id`. */
const SelectBox = object({
  t: literal("select"),
  title: string(),
  description: string(),

  /** If true, the app should refuse to generate until the user has picked a value. */
  required: boolean(),

  /** The key this field's value is sent under in the generation/preview body. */
  id: string(),

  /** If true, more than one of `values` may be selected at once. */
  multiple: boolean(),

  /** The selectable options. `id` is what gets sent, `text` is what the user sees. */
  values: array(
    object({
      id: string(),
      text: string(),
    }),
  ),
});

/**
 * Not an input - a block of server-rendered text that reflects what the user has entered so far.
 * The app should POST the current field values to `live_preview_url` (see getLiveTextPreview)
 * whenever they change and display the text that comes back.
 */
const LiveTextPreview = object({
  t: literal("live_text_preview"),
  live_preview_url: pipe(string(), url()),
});

/** One item in the generation form. Discriminate on `t` to figure out which kind it is. */
const GeneratedStructureItem = union([TextBox, SelectBox, LiveTextPreview]);

/** The whole generation form, in the order the items should be displayed. */
const GenerationStructure = array(GeneratedStructureItem);

/**
 * Fetches the form the app should render for the user, from the generation_endpoint_url we got
 * during the code exchange. Refreshes the token first if it has expired, and if it did,
 * writeCodeExchangeResponse is called with the new body so the caller can persist it.
 */
export async function getGenerationStructure(
  codeExchangeUrl: string,
  codeExchangeResponse: InferOutput<typeof CodeExchangeResponse>,
  writeCodeExchangeResponse: (
    body: InferOutput<typeof CodeExchangeResponse>,
  ) => Promise<void>,
) {
  const resetBody = await refreshIfNeeded(
    codeExchangeUrl,
    codeExchangeResponse,
  );
  if (resetBody) await writeCodeExchangeResponse(resetBody);

  const res = await fetch(
    resetBody?.generation_endpoint_url ||
      codeExchangeResponse.generation_endpoint_url,
    {
      headers: {
        Authorization: `Bearer ${resetBody?.access_token || codeExchangeResponse.access_token}`,
      },
    },
  );
  if (!res.ok) throw new Error("failed to get generation structure");
  return parse(GenerationStructure, await res.json());
}

/**
 * Renders a live_text_preview item by POSTing the current form values (keyed by each item's `id`)
 * to that item's live_preview_url, and returns the text to display. Refreshes the token first if it
 * has expired, and if it did, writeCodeExchangeResponse is called with the new body so the caller
 * can persist it.
 */
export async function getLiveTextPreview(
  codeExchangeUrl: string,
  codeExchangeResponse: InferOutput<typeof CodeExchangeResponse>,
  writeCodeExchangeResponse: (
    body: InferOutput<typeof CodeExchangeResponse>,
  ) => Promise<void>,
  previewUrl: string,
  body: Record<string, string | string[]>,
) {
  const resetBody = await refreshIfNeeded(
    codeExchangeUrl,
    codeExchangeResponse,
  );
  if (resetBody) await writeCodeExchangeResponse(resetBody);

  const res = await fetch(previewUrl, {
    headers: {
      Authorization: `Bearer ${resetBody?.access_token || codeExchangeResponse.access_token}`,
      "Content-Type": "application/json",
    },
    method: "POST",
    body: JSON.stringify(body),
  });
  if (!res.ok) throw new Error("failed to get live preview text");
  return await res.text();
}

/** The shape of the output: an ordered list of parts, each holding some number of image slots. */
const Layout = array(
  object({
    part_name: string(),
    children_count: number(),
  }),
);

/**
 * A terminal status for a single image slot, sent as JSON rather than bytes:
 * `[index]` means that slot is complete, `[index, reason]` means the server refused to generate it.
 */
const GenerationStatus = union([
  strictTuple([number()]),
  strictTuple([number(), string()]),
]);

/** One image slot of the layout, addressed on the wire by its flat index. */
export type GeneratedChild = {
  /** The flat index this slot is addressed by on the wire. */
  index: number;

  /** Every image the generator finished for this slot, in the order they arrived. */
  images: Uint8Array[];

  /** Why the server refused to generate this slot, if it did. */
  rejected: string | null;
};

/** One part of the layout, with its image slots in the order the layout declared them. */
export type GeneratedPart = {
  part_name: string;
  children: GeneratedChild[];
};

/** Joins the chunks buffered for one image into a single array. */
function concatChunks(chunks: Uint8Array[]) {
  const out = new Uint8Array(chunks.reduce((n, chunk) => n + chunk.length, 0));
  let at = 0;
  for (const chunk of chunks) {
    out.set(chunk, at);
    at += chunk.length;
  }
  return out;
}

/**
 * Starts a generation and drains the websocket the generation endpoint hands back.
 *
 * The layout is always the first message, as JSON, and is passed to writeLayout as soon as it lands
 * so the app can put placeholders up before any image data arrives. Every message after that
 * addresses one slot of the layout by its flat index across all of it - with parts of 3 and 2
 * children, index 4 is the second child of the second part.
 *
 * Those messages are either:
 * - bytes, where the top bit of the first byte says whether this is the last chunk of an image and
 *   the low 7 bits are the index. The rest of the array is a chunk, buffered here until the done
 *   bit shows up and we can join them into a finished image.
 * - JSON, either `[index]` (that slot is complete) or `[index, reason]` (the server rejected it).
 *
 * A slot can finish several images before it is marked complete, so the done bit only means "that
 * image is whole", never "that slot is over" - only the JSON status ends a slot. Each finished
 * image is handed straight to writeImage, along with whether its slot is now complete. Every slot
 * ends with exactly one terminal call: writeRejection if the server refused it, otherwise a final
 * writeImage with a null image, so a slot that completes with no image trailing its status still
 * gets an end.
 *
 * Resolves with the whole layout once every slot in it is complete or rejected.
 */
export async function generateImages(
  codeExchangeUrl: string,
  codeExchangeResponse: InferOutput<typeof CodeExchangeResponse>,
  writeCodeExchangeResponse: (
    body: InferOutput<typeof CodeExchangeResponse>,
  ) => Promise<void>,
  body: Record<string, string | string[]>,
  writeLayout: (layout: InferOutput<typeof Layout>) => void,
  writeImage: (
    index: number,
    image: Uint8Array | null,
    complete: boolean,
  ) => void,
  writeRejection: (index: number, reason: string) => void,
) {
  const resetBody = await refreshIfNeeded(
    codeExchangeUrl,
    codeExchangeResponse,
  );
  if (resetBody) await writeCodeExchangeResponse(resetBody);

  const res = await fetch(
    resetBody?.generation_endpoint_url ||
      codeExchangeResponse.generation_endpoint_url,
    {
      headers: {
        Authorization: `Bearer ${resetBody?.access_token || codeExchangeResponse.access_token}`,
        "Content-Type": "application/json",
      },
      method: "POST",
      body: JSON.stringify(body),
    },
  );
  if (!res.ok) throw new Error("failed to get generation websocket url");
  const websocketUrl = new URL(await res.text());
  if (websocketUrl.protocol !== "wss:")
    throw new Error("requires secure websockets");

  return new Promise<GeneratedPart[]>((resolve, reject) => {
    const ws = new WebSocket(websocketUrl);
    ws.binaryType = "arraybuffer";

    /** Built from the first message. Nothing else on the socket means anything until we have it. */
    let parts: GeneratedPart[] | null = null;

    /** Flat index -> where in `parts` that slot lives. Its length is the total slot count. */
    const locations: { part: number; child: number }[] = [];

    /** Chunks held for the image a slot is part way through, keyed by flat index. */
    const chunks = new Map<number, Uint8Array[]>();

    const complete = new Set<number>();
    let settled = false;

    const fail = (err: unknown) => {
      if (settled) return;
      settled = true;
      ws.close();
      reject(err);
    };

    const done = () => {
      settled = true;
      ws.close();
      resolve(parts!);
    };

    const childAt = (index: number) => {
      const at = locations[index];
      if (!at) throw new Error(`index ${index} is outside the layout`);
      return parts![at.part].children[at.child];
    };

    ws.onmessage = (ev) => {
      try {
        if (typeof ev.data === "string") {
          const json = JSON.parse(ev.data);

          // The layout comes first, and only once.
          if (!parts) {
            const layout = parse(Layout, json);
            parts = layout.map((part) => ({
              part_name: part.part_name,
              children: [],
            }));
            for (let part = 0; part < layout.length; part++) {
              for (
                let child = 0;
                child < layout[part].children_count;
                child++
              ) {
                parts[part].children.push({
                  index: locations.length,
                  images: [],
                  rejected: null,
                });
                locations.push({ part, child });
              }
            }
            writeLayout(layout);
            // A layout with no slots in it has nothing left to wait for.
            if (!locations.length) done();
            return;
          }

          const status = parse(GenerationStatus, json);
          const index = status[0];
          const child = childAt(index);
          if (status.length === 2) {
            // Rejected, so the half finished image we were holding is never going to be one.
            child.rejected = status[1];
          }
          chunks.delete(index);

          // The status is what ends a slot, whether or not any image came with it.
          if (!complete.has(index)) {
            complete.add(index);
            if (child.rejected !== null) writeRejection(index, child.rejected);
            else writeImage(index, null, true);
          }
          if (complete.size >= locations.length) done();
          return;
        }

        if (!parts)
          throw new Error("expected the layout before any image data");
        const frame = new Uint8Array(ev.data as ArrayBuffer);
        if (!frame.length) throw new Error("got an empty frame");

        // Top bit of the header byte is the done flag, the low 7 bits are the flat index.
        const isDone = (frame[0] & 0b1000_0000) !== 0;
        const index = frame[0] & 0b0111_1111;
        const child = childAt(index);

        const buffered = chunks.get(index) || [];
        buffered.push(frame.subarray(1));
        if (!isDone) {
          chunks.set(index, buffered);
          return;
        }

        // One whole image, but the slot stays open for more until its status turns up.
        chunks.delete(index);
        const image = concatChunks(buffered);
        child.images.push(image);
        writeImage(index, image, complete.has(index));
      } catch (err) {
        fail(err);
      }
    };

    ws.onerror = () => fail(new Error("generation websocket errored"));
    ws.onclose = () =>
      fail(
        new Error("generation websocket closed before every slot was complete"),
      );
  });
}

// Remote workspace extension (workspace protocol version 1).

export const Id = v.pipe(v.string(), v.minLength(1), v.maxLength(256));
export const Count = v.pipe(v.number(), v.safeInteger(), v.minValue(0));
export const HttpsURL = v.pipe(
  v.string(),
  v.url(),
  v.check((s) => new URL(s).protocol === "https:"),
);
export const WssURL = v.pipe(
  v.string(),
  v.url(),
  v.check((s) => new URL(s).protocol === "wss:"),
);
// MessagePack bin values decode to Uint8Array; strings and numeric arrays are invalid.
export const Bytes = v.instance(Uint8Array);
// west > east represents a boundary crossing the antimeridian.
export const Bounds = v.pipe(
  v.object({
    south: v.pipe(v.number(), v.minValue(-90), v.maxValue(90)),
    north: v.pipe(v.number(), v.minValue(-90), v.maxValue(90)),
    west: v.pipe(v.number(), v.minValue(-180), v.maxValue(180)),
    east: v.pipe(v.number(), v.minValue(-180), v.maxValue(180)),
  }),
  v.check((b) => b.south <= b.north),
);

/** Apply all filters before ranking and pagination; missing EXIF fails date/location filters. */
export const Filters = v.object({
  // Match any listed MIME type (an empty list matches nothing), but all listed tags.
  mime_types: v.optional(v.array(v.string())),
  tags: v.optional(v.array(v.string())),
  edited: v.optional(v.boolean()),
  content: v.optional(v.picklist(["all", "safe", "flagged"])),
  // Inclusive Unix seconds. Rating 0 means unrated.
  captured_after: v.optional(Count),
  captured_before: v.optional(Count),
  min_rating: v.optional(v.pipe(Count, v.maxValue(5))),
  bounds: v.optional(Bounds),
});
export type Filters = v.InferOutput<typeof Filters>;
// Unknown or forbidden IDs must produce an error, never fall back to the whole library.
export const Scope = v.variant("kind", [
  v.object({ kind: v.literal("library") }),
  v.object({
    kind: v.literal("folder"),
    id: Id,
    recursive: v.optional(v.boolean(), true),
  }),
  v.object({ kind: v.literal("bucket"), id: Id }),
]);
// Search only within scope. Break sort ties by asset ID; offset/limit form a live window.
export const AssetQuery = v.object({
  scope: Scope,
  text: v.optional(v.string(), ""),
  filters: v.optional(Filters, {}),
  sort: v.optional(
    v.picklist(["relevance", "name", "captured_desc", "modified_desc"]),
    "name",
  ),
  offset: v.optional(Count, 0),
  limit: v.optional(v.pipe(Count, v.minValue(1), v.maxValue(500)), 100),
});
export type AssetQuery = v.InferInput<typeof AssetQuery>;
export const Folder = v.object({
  id: Id,
  parent_id: v.nullable(Id),
  name: v.string(),
  revision: Count,
});
// Bucket contents are manual members UNION rule matches, deduplicated by asset ID.
// Rules may scope to library/folder, never another bucket (which could create cycles).
export const BucketRule = v.object({
  scope: Scope,
  text: v.string(),
  filters: Filters,
});
export const Bucket = v.object({
  id: Id,
  name: v.string(),
  revision: Count,
  rule: v.nullable(BucketRule),
});
/** A remote file. Its ID also identifies its collaborative document. */
export const Asset = v.object({
  id: Id,
  // Folder membership belongs to the asset, not to an editor session or a bucket.
  // null means an unfiled asset (for example, a local file dropped into a bucket).
  folder_id: v.nullable(Id),
  name: v.string(),
  mime_type: v.string(),
  revision: Count,
  size: Count,
  edited: v.boolean(),
  tags: v.array(v.string()),
  rating: v.pipe(Count, v.maxValue(5)),
  captured_at: v.nullable(Count),
  modified_at: Count,
  thumbnail_url: v.nullable(HttpsURL),
});
export type Asset = v.InferOutput<typeof Asset>;
// Catalogue text searches folder/bucket names; AssetQuery text searches their contents.
export const CatalogueQuery = v.object({
  text: v.optional(v.string(), ""),
  offset: v.optional(Count, 0),
  limit: v.optional(v.pipe(Count, v.minValue(1), v.maxValue(500)), 100),
});
export type CatalogueQuery = v.InferInput<typeof CatalogueQuery>;
export const WatchQuery = v.variant("kind", [
  v.object({ kind: v.literal("folders"), query: CatalogueQuery }),
  v.object({ kind: v.literal("buckets"), query: CatalogueQuery }),
  v.object({ kind: v.literal("assets"), query: AssetQuery }),
]);
export type WatchQuery = v.InferInput<typeof WatchQuery>;
const windowFields = { revision: Count, total: Count, offset: Count };
// Each push replaces the requested window, including empty results and deletions.
// Revision increases within a subscription; total counts all matches before pagination.
export const Snapshot = v.variant("kind", [
  v.object({
    kind: v.literal("folders"),
    ...windowFields,
    items: v.array(Folder),
  }),
  v.object({
    kind: v.literal("buckets"),
    ...windowFields,
    items: v.array(Bucket),
  }),
  v.object({
    kind: v.literal("assets"),
    ...windowFields,
    items: v.array(Asset),
  }),
]);
export type Snapshot = v.InferOutput<typeof Snapshot>;

/** Optional headless document engine extension; transport stays workspace v1. */
export const WorkspaceCapabilities = v.object({
  document_models: v.array(Id),
  formats: v.array(
    v.object({
      id: Id,
      name: v.string(),
      extensions: v.array(v.string()),
      can_export: v.boolean(),
      // A codec can require a runtime library, e.g. HEIC needs libheif/HEVC.
      runtime_requirement: v.nullable(v.string()),
    }),
  ),
  max_frame_bytes: Count,
  max_document_bytes: Count,
  default_edited_export: v.string(),
  original_download: v.boolean(),
});

/**
 * Native image model: schist.image.v1 (all root-map values are Uint8Array).
 * The shared schist-document Rust library is the desktop/provider reference
 * implementation. Importers and exporters use the same built-in codec registry.
 * A provider initializes an existing asset from saved bytes exactly once under
 * the room's join transaction, persists before replying, and never re-seeds an
 * existing native model from an older export. Reserve Yjs client ID 1 for the
 * deterministic initial seed; ordinary participants use other client IDs.
 *
 * Keys:
 * - document/size: MessagePack [width, height, resolution_dpi]
 * - document/title: UTF-8
 * - document/metadata: layerless 1x1 PSD preserving document metadata
 * - document/comps: MessagePack [stable_layer_references, layer_comps]
 * - layer/<id>/placement: MessagePack [parent_id ("root" at top), sibling_rank]
 * - layer/<id>/template: one-layer 1x1 PSD, no raster/mask tiles or child layers
 * - layer/<id>/name: UTF-8
 * - layer/<id>/visible, locked, clipping: one boolean byte
 * - layer/<id>/opacity, fill: little-endian float32
 * - layer/<id>/blend: four-byte PSD blend key
 * - layer/<id>/pixels/<x>/<y>: depth byte 8/16/32, then one full RGBA tile;
 *   multibyte samples are little-endian (Schist tiles are 256x256 pixels)
 * - layer/<id>/mask/<x>/<y>: one full single-channel 8-bit mask tile
 *
 * Initial IDs are deterministic seed/<sibling index>/... paths; new layers use
 * UUIDs. Placement is atomic, ordered by rank then stable ID. Each property and
 * tile merges independently. Concurrent writes to one tile use Yjs resolution,
 * not a per-pixel merge. Local selections, transient tools and undo history are
 * not serialized. The provider validates the materialized model before accepting
 * edits and exports current committed state; it retains original upload bytes.
 * Source-format and export-format capabilities are independent, and retain the
 * reference codec's actual fidelity limits. User-installed codecs must also be
 * installed/trusted on a provider before it advertises support for them.
 *
 * A provider may impose a lower document limit than the 256 MiB wire ceiling.
 * Discover it with workspace.capabilities; oversized edits fail explicitly.
 * This extension does not change the legacy image-generation socket format.
 */

export const Failure = v.object({ code: Id, message: v.string() });
export const ServerMessage = v.variant("type", [
  v.object({ type: v.literal("ready"), protocol: v.literal(1) }),
  v.object({ type: v.literal("result"), id: Id, value: v.unknown() }),
  v.object({ type: v.literal("error"), id: Id, error: Failure }),
  v.object({
    type: v.literal("snapshot"),
    subscription_id: Id,
    snapshot: Snapshot,
  }),
  v.object({
    type: v.literal("watch_error"),
    subscription_id: Id,
    error: Failure,
  }),
  v.object({
    type: v.literal("document_update"),
    document_id: Id,
    update: Bytes,
  }),
  v.object({
    type: v.literal("document_error"),
    document_id: Id,
    error: Failure,
  }),
  v.object({ type: v.literal("auth_expiring") }),
  v.object({ type: v.literal("pong") }),
]);
export type ServerMessage = v.InferOutput<typeof ServerMessage>;
type ServerReply = Extract<ServerMessage, { type: "result" | "error" }>;
type Failure = v.InferOutput<typeof Failure>;

// Every workspace WebSocket message is one MessagePack map in a binary frame.
// Field names remain strings; updates/state vectors use bin values (Uint8Array).
// Omit absent optional fields; use nil only where the schema explicitly allows null.
// Requests/replies and pushes share one connection. Request IDs correlate replies;
// subscription/document IDs route updates independently of requests in flight.
type ClientMessage =
  | { type: "hello"; protocol: 1; access_token: string }
  | { type: "request"; id: string; method: string; params: unknown }
  | { type: "subscribe"; subscription_id: string; query: WatchQuery }
  | { type: "unsubscribe"; subscription_id: string }
  | { type: "ping" };

export class RemoteError extends Error {
  constructor(
    public readonly code: string,
    message: string,
  ) {
    super(message);
  }
}

type ConnectionState = "connecting" | "connected" | "disconnected" | "closed";

type PendingRequest = {
  resolve: (value: unknown) => void;
  reject: (error: Error) => void;
  timer: ReturnType<typeof setTimeout>;
};

export type RemoteWorkspaceOptions = {
  /** WASM/browser builds must enable the official origin and endpoint restrictions. */
  browserCloud?: boolean;
  codeExchangeUrl: string;
  credentials: InferOutput<typeof CodeExchangeResponse>;
  writeCredentials: (
    body: InferOutput<typeof CodeExchangeResponse>,
  ) => Promise<void>;
  onError: (error: Error) => void;
  onState?: (state: ConnectionState) => void;
  /** Injectable for native hosts/tests; production must preserve TLS validation. */
  socketFactory?: (url: string) => WebSocket;
};

type Watch = {
  query: WatchQuery;
  revision: number;
  receive: (snapshot: Snapshot) => void;
  error: (error: Error) => void;
};

type DocumentBinding = {
  doc: Y.Doc;
  joined: boolean;
  epoch: number;
  pending: Uint8Array[];
  sending: Promise<void> | null;
  joining: Promise<void> | null;
  listener: (update: Uint8Array, origin: unknown) => void;
  error: (error: Error) => void;
};

export type BucketDrop =
  | { kind: "asset"; id: string }
  | { kind: "folder"; id: string; recursive?: boolean }
  | { kind: "local"; name: string; bytes: Blob; relative_path?: string };

const MutationResult = v.object({ id: Id, revision: Count });
const UploadTicket = v.object({ upload_id: Id, put_url: HttpsURL });
const JoinResult = v.object({ update: Bytes, state_vector: Bytes });
const Ack = v.object({});
const MAX_FRAME_BYTES = 256 * 1024 * 1024;
const HANDSHAKE_TIMEOUT_MS = 15_000;
const REQUEST_TIMEOUT_MS = 30_000;
const HEARTBEAT_INTERVAL_MS = 20_000;
const HEARTBEAT_TIMEOUT_MS = 60_000;
const TOKEN_REFRESH_MARGIN_SECONDS = 30;

function asError(error: unknown): Error {
  return error instanceof Error ? error : new Error(String(error));
}

/**
 * One instance per signed-in account: folders, buckets and editors share its socket.
 * workspace protocol version 1 is independent of the original schist_spec_version.
 * Uses valibot, yjs and @msgpack/msgpack. Legacy generation keeps its existing wire format.
 */
export class RemoteWorkspace {
  private credentials: InferOutput<typeof CodeExchangeResponse>;
  private socket: WebSocket | null = null;
  private ready = false;
  private stopped = false;
  private retry = 0;
  private refreshRequired = false;
  private retryTimer?: ReturnType<typeof setTimeout>;
  private heartbeat?: ReturnType<typeof setInterval>;
  private openTimer?: ReturnType<typeof setTimeout>;
  private lastMessage = Date.now();
  private pending = new Map<string, PendingRequest>();
  private watches = new Map<string, Watch>();
  private documents = new Map<string, DocumentBinding>();

  constructor(private readonly options: RemoteWorkspaceOptions) {
    if (options.browserCloud) {
      officialBrowser();
      officialCloudUrl(options.codeExchangeUrl);
      browserCredentials(options.credentials);
      if (options.socketFactory) throw new Error("Browser cloud uses the browser WebSocket implementation");
    }
    this.credentials = options.credentials;
    if (!this.credentials.workspace_websocket_url)
      throw new Error("provider has no remote workspace endpoint");
    v.parse(WssURL, this.credentials.workspace_websocket_url);
    void this.connect();
  }

  get connected() {
    return this.ready;
  }

  private report(error: unknown) {
    try {
      this.options.onError(asError(error));
    } catch {
      // Reporting a UI error must not interrupt transport cleanup or report recursively.
    }
  }

  private notify(callback: () => void) {
    try {
      callback();
    } catch (error) {
      // User callbacks are isolated from protocol errors: a broken view need not disconnect.
      this.report(error);
    }
  }

  private state(state: ConnectionState) {
    this.notify(() => this.options.onState?.(state));
  }

  private async connect() {
    if (this.stopped) return;
    this.state("connecting");

    try {
      await this.refreshCredentials();
      if (this.stopped) return;

      const url = v.parse(WssURL, this.credentials.workspace_websocket_url);
      if (this.options.browserCloud) officialCloudUrl(url, true);
      const createSocket =
        this.options.socketFactory ?? ((url) => new WebSocket(url));
      this.attachSocket(createSocket(url));
    } catch (error) {
      this.report(error);
      this.scheduleReconnect();
    }
  }

  private async refreshCredentials() {
    const refreshAt = Date.now() / 1000 + TOKEN_REFRESH_MARGIN_SECONDS;
    if (!this.refreshRequired && this.credentials.expires_at > refreshAt)
      return;

    const fresh = await performTokenRefresh(
      this.options.codeExchangeUrl,
      this.credentials.refresh_token,
    );
    // Refresh tokens can rotate. Keep the new token even if persistence fails.
    if (this.options.browserCloud) browserCredentials(fresh);
    this.credentials = fresh;
    this.refreshRequired = false;
    await this.options.writeCredentials(fresh);
  }

  private attachSocket(socket: WebSocket) {
    // Decode synchronously in arrival order; Blob conversion could reorder messages.
    socket.binaryType = "arraybuffer";
    this.socket = socket;
    this.openTimer = setTimeout(
      () => this.disconnect(new Error("workspace handshake timed out")),
      HANDSHAKE_TIMEOUT_MS,
    );

    // These callbacks only dispatch. A single boundary catches malformed frames,
    // socket send failures and late events from a superseded connection.
    socket.onopen = () =>
      this.handleSocketEvent(socket, () => this.authenticate());
    socket.onmessage = (event) =>
      this.handleSocketEvent(socket, () => this.receive(event.data));
    socket.onerror = () =>
      this.handleSocketEvent(socket, () => {
        throw new Error("workspace socket failed");
      });
    socket.onclose = (event) => this.handleClose(socket, event.code);
  }

  private handleSocketEvent(socket: WebSocket, handle: () => void) {
    if (this.socket !== socket || this.stopped) return;
    try {
      handle();
    } catch (error) {
      this.disconnect(error);
    }
  }

  private authenticate() {
    // Before ready, only hello is permitted. Authenticate in the frame rather than
    // putting a bearer token in a URL that proxies or browser history could retain.
    // Server authenticates before ready and checks permissions on every request,
    // subscription and document update, including after access is revoked.
    this.send({
      type: "hello",
      protocol: 1,
      access_token: this.credentials.access_token,
    });
  }

  private receive(data: unknown) {
    // A frame queued before disconnect must not reopen a connection being closed.
    if (this.socket?.readyState !== WebSocket.OPEN) return;
    if (!(data instanceof ArrayBuffer) || data.byteLength > MAX_FRAME_BYTES) {
      throw new Error("invalid workspace frame");
    }
    // decode consumes exactly one value, rejecting truncated or trailing data.
    // Bound advertised lengths too, before a malformed packet can allocate a huge container.
    const message = v.parse(
      ServerMessage,
      decode(new Uint8Array(data), {
        maxStrLength: MAX_FRAME_BYTES,
        maxBinLength: MAX_FRAME_BYTES,
        maxArrayLength: MAX_FRAME_BYTES,
        maxMapLength: MAX_FRAME_BYTES,
        maxExtLength: 0,
      }),
    );
    this.lastMessage = Date.now();

    if (message.type !== "ready" && !this.ready) {
      throw new Error("expected workspace ready");
    }

    switch (message.type) {
      case "ready":
        return this.handleReady();
      case "result":
      case "error":
        return this.handleReply(message);
      case "snapshot":
        return this.handleSnapshot(message.subscription_id, message.snapshot);
      case "watch_error":
        return this.handleWatchError(message.subscription_id, message.error);
      case "document_update":
        return this.handleDocumentUpdate(message.document_id, message.update);
      case "document_error":
        return this.handleDocumentError(message.document_id, message.error);
      case "auth_expiring":
        this.refreshRequired = true;
        return this.disconnect(new Error("workspace token renewal"));
      case "pong":
        return; // Receipt already refreshed lastMessage for the heartbeat.
    }
  }

  private handleReady() {
    if (this.ready) throw new Error("duplicate workspace handshake");
    clearTimeout(this.openTimer);
    this.ready = true;
    this.retry = 0;
    this.heartbeat = setInterval(
      () => this.checkHeartbeat(),
      HEARTBEAT_INTERVAL_MS,
    );

    // Watches restart with full snapshots. Documents instead exchange state vectors
    // so both sides recover missing edits, including changes made while offline.
    for (const [id, watch] of this.watches) {
      watch.revision = -1;
      this.send({ type: "subscribe", subscription_id: id, query: watch.query });
    }
    for (const [assetId, binding] of this.documents)
      void this.join(assetId, binding);
    this.state("connected");
  }

  private checkHeartbeat() {
    if (!this.socket || !this.ready) return;
    this.handleSocketEvent(this.socket, () => {
      if (Date.now() - this.lastMessage > HEARTBEAT_TIMEOUT_MS) {
        throw new Error("workspace heartbeat timed out");
      }
      this.send({ type: "ping" });
    });
  }

  private handleReply(message: ServerReply) {
    const request = this.pending.get(message.id);
    if (!request) return; // A late reply cannot resolve a request that already timed out.

    this.pending.delete(message.id);
    clearTimeout(request.timer);
    if (message.type === "error") {
      request.reject(
        new RemoteError(message.error.code, message.error.message),
      );
    } else {
      request.resolve(message.value);
    }
  }

  private handleSnapshot(subscriptionId: string, snapshot: Snapshot) {
    const watch = this.watches.get(subscriptionId);
    // Query replacement uses a new ID; ignore cancelled watches and stale results.
    if (!watch || snapshot.revision <= watch.revision) return;
    if (watch.query.kind !== snapshot.kind)
      throw new Error("wrong subscription snapshot kind");

    watch.revision = snapshot.revision;
    this.notify(() => watch.receive(snapshot));
  }

  private handleWatchError(subscriptionId: string, error: Failure) {
    const watch = this.watches.get(subscriptionId);
    // A rejected/revoked subscription is terminal; reconnect must not resurrect it.
    this.watches.delete(subscriptionId);
    this.notify(() => watch?.error(new RemoteError(error.code, error.message)));
  }

  private handleDocumentUpdate(assetId: string, update: Uint8Array) {
    const binding = this.documents.get(assetId);
    if (!binding) return;

    // Mark the transaction as remote so the local listener does not echo it back.
    Y.applyUpdate(binding.doc, update, this);
  }

  private handleDocumentError(assetId: string, error: Failure) {
    const binding = this.documents.get(assetId);
    if (!binding) return;

    this.detachDocument(assetId, binding);
    this.notify(() =>
      binding.error(new RemoteError(error.code, error.message)),
    );
  }

  private handleClose(socket: WebSocket, code: number) {
    if (this.socket !== socket) return;
    if (code === 4401) this.refreshRequired = true;
    this.socket = null;
    this.reset(
      new Error(
        "workspace disconnected; outstanding mutation outcomes may be unknown",
      ),
    );

    // Reconnect only after close, including token renewal: never overlap sockets.
    this.scheduleReconnect();
  }

  private scheduleReconnect() {
    if (this.stopped || this.retryTimer) return;
    this.state("disconnected");
    // Cap exponential backoff and add jitter so reconnecting clients spread out.
    const delay =
      Math.min(30_000, 500 * 2 ** Math.min(this.retry++, 6)) *
      (0.75 + Math.random() / 2);
    this.retryTimer = setTimeout(() => {
      this.retryTimer = undefined;
      void this.connect();
    }, delay);
  }

  private reset(error: Error) {
    this.ready = false;
    clearTimeout(this.openTimer);
    clearInterval(this.heartbeat);
    for (const request of this.pending.values()) {
      clearTimeout(request.timer);
      request.reject(error);
    }
    this.pending.clear();
    for (const binding of this.documents.values()) {
      binding.joined = false;
      binding.epoch++;
      binding.sending = null;
      binding.joining = null;
    }
  }

  private disconnect(error: unknown) {
    this.report(error);
    this.reset(asError(error));
    if (this.socket) this.socket.close();
    else this.scheduleReconnect();
  }

  private send(message: ClientMessage) {
    if (!this.socket || this.socket.readyState !== WebSocket.OPEN)
      throw new Error("workspace is not connected");
    // Preserve Uint8Array as MessagePack bin, and keep optional fields absent.
    const frame = encode(message, { ignoreUndefined: true });
    if (frame.byteLength > MAX_FRAME_BYTES)
      throw new Error(
        "workspace frame exceeds 256 MiB; split document transactions",
      );
    if (this.socket.bufferedAmount > MAX_FRAME_BYTES)
      throw new Error("workspace send buffer is full");
    this.socket.send(frame);
  }

  // Mutations carry a stable mutation_id in params, distinct from this request ID.
  // The server persists/replays identical results per principal + mutation_id and
  // rejects reuse with different payloads. Never blindly replay a timed-out mutation.
  private request(method: string, params: unknown): Promise<unknown> {
    if (!this.ready)
      return Promise.reject(new Error("workspace is not connected"));
    const id = crypto.randomUUID();
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.pending.delete(id);
        reject(
          new RemoteError(
            "timeout",
            `${method} timed out; mutation outcome may be unknown`,
          ),
        );
      }, REQUEST_TIMEOUT_MS);
      this.pending.set(id, { resolve, reject, timer });
      try {
        this.send({ type: "request", id, method, params });
      } catch (err) {
        clearTimeout(timer);
        this.pending.delete(id);
        reject(err);
      }
    });
  }

  /** Watch a live result window; close this watch and create another to change its query. */
  watch(
    query: WatchQuery,
    receive: (snapshot: Snapshot) => void,
    onError: (error: Error) => void = (e) => this.report(e),
  ) {
    if (this.stopped) throw new Error("workspace is closed");
    const id = crypto.randomUUID();
    const parsed = v.parse(WatchQuery, query);
    // The server must register and snapshot atomically, then reevaluate membership,
    // filters, ranking and paging on changes. It may coalesce snapshots for slow readers,
    // but must not drop document updates; disconnect if the outgoing queue overflows.
    this.watches.set(id, {
      query: parsed,
      receive,
      error: onError,
      revision: -1,
    });
    if (this.ready) {
      try {
        this.send({ type: "subscribe", subscription_id: id, query: parsed });
      } catch (err) {
        this.disconnect(err);
      }
    }
    return () => {
      if (!this.watches.delete(id) || !this.ready) return;
      try {
        this.send({ type: "unsubscribe", subscription_id: id });
      } catch (err) {
        this.disconnect(err);
      }
    };
  }

  watchFolders(query: CatalogueQuery, receive: (snapshot: Snapshot) => void) {
    return this.watch({ kind: "folders", query }, receive);
  }
  watchBuckets(query: CatalogueQuery, receive: (snapshot: Snapshot) => void) {
    return this.watch({ kind: "buckets", query }, receive);
  }
  /** Empty text lists assets; nonempty text searches the same live, filtered scope. */
  watchAssets(query: AssetQuery, receive: (snapshot: Snapshot) => void) {
    return this.watch({ kind: "assets", query }, receive);
  }

  async createBucket(
    name: string,
    rule: v.InferInput<typeof BucketRule> | null = null,
    mutationId: string = crypto.randomUUID(),
  ) {
    if (rule?.scope.kind === "bucket")
      throw new Error("a bucket rule cannot scope to another bucket");
    return v.parse(
      Bucket,
      await this.request("bucket.create", {
        name,
        rule: v.parse(v.nullable(BucketRule), rule),
        mutation_id: v.parse(Id, mutationId),
      }),
    );
  }
  async updateBucket(
    id: string,
    revision: number,
    name: string,
    rule: v.InferInput<typeof BucketRule> | null,
    mutationId: string = crypto.randomUUID(),
  ) {
    if (rule?.scope.kind === "bucket")
      throw new Error("a bucket rule cannot scope to another bucket");
    return v.parse(
      Bucket,
      await this.request("bucket.update", {
        id: v.parse(Id, id),
        revision: v.parse(Count, revision),
        name,
        rule: v.parse(v.nullable(BucketRule), rule),
        mutation_id: v.parse(Id, mutationId),
      }),
    );
  }
  /** Delete the bucket and its membership, never the original assets. */
  async deleteBucket(
    id: string,
    revision: number,
    mutationId: string = crypto.randomUUID(),
  ) {
    v.parse(
      Ack,
      await this.request("bucket.delete", {
        id: v.parse(Id, id),
        revision: v.parse(Count, revision),
        mutation_id: v.parse(Id, mutationId),
      }),
    );
  }
  async createFolder(
    name: string,
    parentId: string | null = null,
    mutationId: string = crypto.randomUUID(),
  ) {
    return v.parse(
      Folder,
      await this.request("folder.create", {
        name,
        parent_id: v.parse(v.nullable(Id), parentId),
        mutation_id: v.parse(Id, mutationId),
      }),
    );
  }
  async renameFolder(
    id: string,
    revision: number,
    name: string,
    mutationId: string = crypto.randomUUID(),
  ) {
    return v.parse(
      Folder,
      await this.request("folder.update", {
        id: v.parse(Id, id),
        revision: v.parse(Count, revision),
        name,
        mutation_id: v.parse(Id, mutationId),
      }),
    );
  }
  /** The server rejects deletion of a nonempty folder. */
  async deleteFolder(
    id: string,
    revision: number,
    mutationId: string = crypto.randomUUID(),
  ) {
    v.parse(
      Ack,
      await this.request("folder.delete", {
        id: v.parse(Id, id),
        revision: v.parse(Count, revision),
        mutation_id: v.parse(Id, mutationId),
      }),
    );
  }

  /**
   * Add membership without moving originals. Expand remote folders atomically at commit.
   * IDs belong to this account; cross-provider items must be downloaded and uploaded.
   */
  async addToBucket(
    id: string,
    items: Array<
      | { kind: "asset"; id: string }
      | { kind: "folder"; id: string; recursive?: boolean }
    >,
    mutationId: string = crypto.randomUUID(),
  ) {
    const Item = v.variant("kind", [
      v.object({ kind: v.literal("asset"), id: Id }),
      v.object({
        kind: v.literal("folder"),
        id: Id,
        recursive: v.optional(v.boolean(), true),
      }),
    ]);
    return v.parse(
      MutationResult,
      await this.request("bucket.add", {
        id: v.parse(Id, id),
        items: v.parse(v.array(Item), items),
        mutation_id: v.parse(Id, mutationId),
      }),
    );
  }
  /** Removes manual membership; a matching smart rule can still include the asset. */
  async removeFromBucket(
    id: string,
    assetIds: string[],
    mutationId: string = crypto.randomUUID(),
  ) {
    return v.parse(
      MutationResult,
      await this.request("bucket.remove", {
        id: v.parse(Id, id),
        asset_ids: v.parse(v.array(Id), assetIds),
        mutation_id: v.parse(Id, mutationId),
      }),
    );
  }

  /**
   * Import a new file into target.folder_id, or save an existing asset with asset_id
   * + revision. New files without folder_id are unfiled; saves without it keep their
   * current folder. relative_path describes a local import path, not a remote folder ID.
   * prepare/commit IDs derive from mutationId, so retrying the same call is safe.
   * Server keeps a ticket usable for identical retries until its documented expiry.
   */
  async uploadAsset(
    file: { name: string; bytes: Blob; relative_path?: string },
    target: { folder_id?: string; asset_id?: string; revision?: number } = {},
    mutationId: string = crypto.randomUUID(),
    signal?: AbortSignal,
  ) {
    v.parse(Id, mutationId);
    if ((target.asset_id === undefined) !== (target.revision === undefined))
      throw new Error("a save requires asset_id and revision together");
    // The server must enforce the same path rules; a client check is only early feedback.
    if (
      file.relative_path !== undefined &&
      (file.relative_path.includes("\\") ||
        file.relative_path.includes("\0") ||
        file.relative_path
          .split("/")
          .some((p) => !p || p === "." || p === "..") ||
        /^[A-Za-z]:/.test(file.relative_path))
    ) {
      throw new Error("relative_path must be a safe relative path");
    }
    const parsedTarget = v.parse(
      v.object({
        folder_id: v.optional(Id),
        asset_id: v.optional(Id),
        revision: v.optional(Count),
      }),
      target,
    );
    signal?.throwIfAborted();
    const ticket = v.parse(
      UploadTicket,
      await this.request("asset.prepare_upload", {
        ...parsedTarget,
        name: file.name,
        relative_path: file.relative_path,
        mime_type: file.bytes.type || "application/octet-stream",
        size: file.bytes.size,
        mutation_id: `${mutationId}:prepare`,
      }),
    );
    // Raw bytes go to a short-lived HTTPS upload ticket without an account token.
    // Collaboration (including paint data) always travels through document.update.
    if (this.options.browserCloud) officialUploadUrl(ticket.put_url);
    const response = await fetch(ticket.put_url, {
      method: "PUT",
      body: file.bytes,
      signal,
      credentials: "omit",
      redirect: "error",
      headers: {
        "Content-Type": file.bytes.type || "application/octet-stream",
      },
    });
    if (!response.ok) throw new Error(`upload failed (${response.status})`);
    signal?.throwIfAborted();
    // Server verifies ownership/bytes/size and rechecks the expected revision before
    // publishing. Reject stale saves with "conflict" and saves to an active collab room.
    // Uncommitted uploads expire. All mutations must commit before replying/publishing.
    return v.parse(
      Asset,
      await this.request("asset.commit_upload", {
        upload_id: ticket.upload_id,
        mutation_id: `${mutationId}:commit`,
      }),
    );
  }

  /**
   * Drop a selection of local files, remote assets and/or whole remote folders.
   * Enumerate local folders as local items with relative_path (File.webkitRelativePath).
   * Uploads finish first; membership is then added atomically. On failure, successful
   * uploads may exist in the remote library; retry the same items with the same ID.
   */
  async dropIntoBucket(
    bucketId: string,
    items: BucketDrop[],
    mutationId: string = crypto.randomUUID(),
    signal?: AbortSignal,
  ) {
    v.parse(Id, bucketId);
    v.parse(Id, mutationId);
    const remote: Array<
      | { kind: "asset"; id: string }
      | { kind: "folder"; id: string; recursive?: boolean }
    > = [];
    for (const [index, item] of items.entries()) {
      signal?.throwIfAborted();
      if (item.kind === "local") {
        const asset = await this.uploadAsset(
          item,
          {},
          `${mutationId}:${index}`,
          signal,
        );
        remote.push({ kind: "asset", id: asset.id });
      } else remote.push(item);
    }
    signal?.throwIfAborted();
    return this.addToBucket(bucketId, remote, mutationId);
  }

  /** Optional extension. Older providers may return method_not_found. */
  async getCapabilities() {
    return v.parse(WorkspaceCapabilities, await this.request('workspace.capabilities', {}));
  }

  /**
   * Download a revision-consistent export through a short-lived HTTPS URL.
   * Omit format for the provider's current editable export (PSD/PSB for the
   * schist.image.v1 model); use "original" for immutable imported bytes, or an
   * advertised export extension such as "png". Import-only formats remain fully
   * editable in collaboration; they do not imply an encoder for that source format.
   * Decode by file signature and honor Content-Type/Content-Disposition: an edited
   * camera raw or HEIC can be returned as a layered PSD, not its original format.
   * A 409 on GET means the asset advanced before the ticket was used: request a
   * fresh download URL. Bytes served must correspond to the returned revision.
   */
  async downloadAsset(id: string, format?: string) {
    const ticket = v.parse(
      v.object({ url: HttpsURL, revision: Count }),
      await this.request('asset.download', {
        id: v.parse(Id, id),
        ...(format === undefined
          ? {}
          : {
              format: v.parse(v.pipe(v.string(), v.regex(/^[a-z0-9]+$/), v.maxLength(20)), format),
            }),
      }),
    );
    if (this.options.browserCloud) officialCloudUrl(ticket.url);
    return ticket;
  }

  /**
   * Open collaboration for an EXISTING asset: assetId is Asset.id, and the server
   * resolves its folder through Asset.folder_id. Opening never creates or moves a file.
   * To import into a folder, uploadAsset(file, { folder_id }) first, then open its ID.
   * Local/offline changes stay in the caller-owned Y.Doc; persist it to survive exit.
   * onError surfaces unsaved/rejected edits. close() does not save or destroy the doc.
   */
  openDocument(
    assetId: string,
    doc: Y.Doc = new Y.Doc(),
    onError: (error: Error) => void = (e) => this.report(e),
  ) {
    v.parse(Id, assetId);
    if (this.stopped) throw new Error("workspace is closed");
    if (this.documents.has(assetId))
      throw new Error("document already bound to this workspace");
    if ([...this.documents.values()].some((b) => b.doc === doc))
      throw new Error("a Y.Doc cannot represent two remote documents");
    const binding: DocumentBinding = {
      doc,
      joined: false,
      epoch: 0,
      pending: [],
      sending: null,
      joining: null,
      error: onError,
      listener: (update, origin) => {
        // Applying a remote update uses this workspace as its transaction origin.
        if (origin === this) return;
        binding.pending.push(update);
        // Coalesce local transactions while offline or waiting for an acknowledgement.
        if (binding.pending.length > 16)
          binding.pending = [Y.mergeUpdates(binding.pending)];
        void this.pump(assetId, binding);
      },
    };
    this.documents.set(assetId, binding);
    doc.on("update", binding.listener);
    if (this.ready) void this.join(assetId, binding);
    return {
      assetId,
      doc,
      /** Await the initial join (or retry a failed join) before editing/flush. */
      synchronize: async () => {
        if (!this.ready || this.documents.get(assetId) !== binding)
          throw new Error("workspace/document is not connected");
        await this.join(assetId, binding);
        if (!binding.joined) throw new Error("document synchronization failed");
      },
      /** Resolve after queued edits are acknowledged; reject if not joined/connected. */
      flush: async () => {
        do {
          if (!binding.joined || this.documents.get(assetId) !== binding)
            throw new Error("document is not synchronized");
          await this.pump(assetId, binding);
        } while (binding.pending.length || binding.sending);
        if (!binding.joined)
          throw new Error("document disconnected before flush completed");
      },
      close: () => {
        if (this.documents.get(assetId) !== binding) return;
        this.detachDocument(assetId, binding);
        if (this.ready)
          void this.request("document.leave", { document_id: assetId }).catch(
            (e) => this.report(e),
          );
      },
    };
  }

  private detachDocument(assetId: string, binding: DocumentBinding) {
    this.documents.delete(assetId);
    binding.doc.off("update", binding.listener);
    binding.joined = false;
    binding.epoch++;
    binding.sending = null;
  }

  private join(assetId: string, binding: DocumentBinding): Promise<void> {
    if (binding.joining) return binding.joining;
    if (binding.joined) return Promise.resolve();
    // Disconnect/detach advances the epoch, preventing late replies from changing
    // a replacement binding or consuming updates queued for a newer connection.
    const epoch = ++binding.epoch;
    const joining = this.performJoin(assetId, binding, epoch).finally(() => {
      if (binding.epoch === epoch) binding.joining = null;
    });
    binding.joining = joining;
    return joining;
  }

  private async performJoin(
    assetId: string,
    binding: DocumentBinding,
    epoch: number,
  ) {
    try {
      // document_id is the asset ID, not a folder ID or a new session identifier.
      // Server must resolve an existing authorized asset and initialize its shared
      // model from the saved edit once. Atomically join and return a diff + vector
      // from the same server state, so edits cannot fall into a snapshot/subscribe gap.
      const result = v.parse(
        JoinResult,
        await this.request("document.join", {
          document_id: assetId,
          state_vector: Y.encodeStateVector(binding.doc),
        }),
      );
      if (this.documents.get(assetId) !== binding || binding.epoch !== epoch)
        return;
      Y.applyUpdate(binding.doc, result.update, this);
      // Includes local edits made DURING join, as well as edits made offline.
      binding.pending = [
        Y.encodeStateAsUpdate(binding.doc, result.state_vector),
      ];
      binding.joined = true;
      await this.pump(assetId, binding);
    } catch (err) {
      if (binding.epoch !== epoch) return;
      binding.joined = false;
      this.notify(() => binding.error(asError(err)));
    }
  }

  private pump(assetId: string, binding: DocumentBinding): Promise<void> {
    if (binding.sending) return binding.sending;
    if (!this.ready || !binding.joined || !binding.pending.length)
      return Promise.resolve();
    const epoch = binding.epoch;
    // Only one update is in flight per document. Keep the batch until acknowledged;
    // server must persist/apply it before replying and broadcasting to joined peers.
    // Yjs v1 updates are idempotent: https://docs.yjs.dev/api/document-updates
    const update = Y.mergeUpdates(binding.pending);
    binding.pending = [];
    const sending = this.request("document.update", {
      document_id: assetId,
      update,
    })
      .then((result) => {
        v.parse(Ack, result);
      })
      .catch((err) => {
        if (binding.epoch !== epoch) return;
        binding.pending.unshift(update);
        binding.joined = false; // Explicit error; no busy retry loop for forbidden/invalid updates.
        this.notify(() => binding.error(asError(err)));
      })
      .finally(() => {
        if (binding.epoch !== epoch) return;
        binding.sending = null;
        if (binding.joined && binding.pending.length)
          void this.pump(assetId, binding);
      });
    binding.sending = sending;
    return sending;
  }

  close() {
    if (this.stopped) return;
    this.stopped = true;
    clearTimeout(this.retryTimer);
    this.reset(new Error("workspace closed"));
    for (const [id, binding] of this.documents)
      this.detachDocument(id, binding);
    this.watches.clear();
    this.socket?.close();
    this.state("closed");
  }
}

/**
 * Example (all these views and documents use the same socket):
 *
 * const remote = new RemoteWorkspace({ codeExchangeUrl, credentials,
 *   writeCredentials: persistLogin, onError: showError });
 * const stopFolders = remote.watchFolders({}, renderFolders);
 * const stopBuckets = remote.watchBuckets({ text: "holiday" }, renderBuckets);
 * let stopPhotos = remote.watchAssets({
 *   scope: { kind: "folder", id: "folder-1" }, text: "sunset",
 *   filters: { content: "safe", edited: true }, sort: "relevance",
 * }, renderPhotos);
 * // Changing search/scope/filters: cancel the old watch, start a new one.
 * stopPhotos();
 * stopPhotos = remote.watchAssets({
 *   scope: { kind: "bucket", id: "bucket-1" }, text: "beach",
 *   filters: { min_rating: 3 },
 * }, renderPhotos);
 * // Once onState reports connected, mutations can be sent.
 * await remote.dropIntoBucket("bucket-1", [
 *   { kind: "folder", id: "folder-1", recursive: true },
 *   { kind: "local", name: file.name, bytes: file, relative_path: file.webkitRelativePath || undefined },
 * ]);
 * // Folder association is set on import. Later, a gallery click can pass any
 * // existing Asset.id directly to openDocument, including assets shown in a bucket.
 * const asset = await remote.uploadAsset(
 *   { name: file.name, bytes: file }, { folder_id: "folder-1" },
 * );
 * const editing = remote.openDocument(asset.id);
 * await editing.synchronize();
 * editing.doc.getMap("properties").set("title", "Summer");
 * // The editor maps operations to Yjs types: layer order -> Y.Array of stable IDs,
 * // layer properties -> nested Y.Maps, paint tiles -> Uint8Array data at stable tile keys.
 * // Concurrent writes to a single key use Yjs conflict resolution; this is not a
 * // PSD merge algorithm. The Rust editor binding comes later.
 * await editing.flush();
 * editing.close();
 * stopFolders(); stopBuckets(); stopPhotos(); remote.close();
 */