/** Local loopback client. Writes are never automatically retried. */
export type Scalar = null | boolean | number | string;
export type JsonPredicate =
  | {op: "eq" | "ne" | "lt" | "le" | "gt" | "ge"; path: string[]; value: Scalar}
  | {op: "exists"; path: string[]; present: boolean}
  | {op: "type"; path: string[]; kind: "null" | "boolean" | "number" | "string" | "array" | "object"}
  | {op: "prefix"; path: string[]; value: string}
  | {op: "all" | "any"; predicates: JsonPredicate[]}
  | {op: "not"; predicate: JsonPredicate};
export type Query = {key_prefix: string; predicate: "any" | "equals" | "prefix"; value: string}
  | {key_prefix: string; predicate: "json"; value: ""; json: JsonPredicate};
export interface Row {key: string; value: string}
export interface Page {position: string; rows: Row[]; next: string | null; explain: {plan: string; position: string; scanned_records: number; scanned_rows: number; matched_rows: number; returned_rows: number; elapsed_micros: number}}
export interface Session {id: string; instance: string; token: string; position: string; rows: Row[]}
export interface Change {key: string; before: string | null; after: string | null}
export interface StreamPage {commits: {position: string; changes: Change[]}[]; token: string; position: string; head: string; floor: string}
export interface RevisionRead {value: string | null; revision: string; position: string}
export interface Receipt {position: string; durable: boolean; wrote: boolean}
export class ApiError extends Error {
  status: number; body: unknown;
  constructor(status: number, body: unknown) {super(`Polybase HTTP ${status}: ${JSON.stringify(body)}`);this.status=status;this.body=body;}
}
export const hex = (value: Uint8Array | string): string => [...(typeof value === "string" ? new TextEncoder().encode(value) : value)].map(b=>b.toString(16).padStart(2,"0")).join("");
export function unhex(value: string): Uint8Array {
  if (!/^(?:[0-9a-f]{2})*$/.test(value)) throw new Error("invalid hex");
  return Uint8Array.from(value.match(/../g) || [], b=>Number.parseInt(b,16));
}
function safeNumbers(value: unknown, depth = 0): void {
  if (depth > 64) throw new Error("request nesting exceeds 64");
  if (typeof value === "number" && (!Number.isFinite(value) || (Number.isInteger(value) && !Number.isSafeInteger(value)))) throw new Error("unsafe JSON number; use strings for exact large integers");
  if (value && typeof value === "object") for (const child of Object.values(value)) safeNumbers(child,depth+1);
}
export class Client {
  readonly base: string;
  constructor(base = "http://127.0.0.1:7340") {
    if (!/^http:\/\/127\.0\.0\.1:[1-9][0-9]*$/.test(base) || Number(new URL(base).port)>65535) throw new Error("use an IPv4 loopback origin with an explicit port");
    this.base=base;
  }
  async call<T>(route: string, body: unknown, signal?: AbortSignal): Promise<T> {
    if (!/^[a-z-]+$/.test(route)) throw new Error("invalid route");
    safeNumbers(body);
    const payload=JSON.stringify(body);
    if (new TextEncoder().encode(payload).length>128*1024) throw new Error("request exceeds 128 KiB");
    const response=await fetch(`${this.base}/api/${route}`,{method:"POST",headers:{"Content-Type":"application/json",Origin:this.base},body:payload,signal:signal ? AbortSignal.any([signal,AbortSignal.timeout(5000)]) : AbortSignal.timeout(5000),redirect:"error"});
    const reader=response.body?.getReader();
    if (!reader) throw new Error("missing response body");
    const chunks: Uint8Array[]=[];let size=0;
    try {
      while (true) {const {done,value}=await reader.read();if(done) break;size+=value.length;if(size>8*1024*1024) throw new Error("response exceeds 8 MiB");chunks.push(value);}
    } finally {await reader.cancel();reader.releaseLock();}
    const all=new Uint8Array(size);let offset=0;for(const chunk of chunks){all.set(chunk,offset);offset+=chunk.length;}
    const result=JSON.parse(new TextDecoder("utf-8",{fatal:true}).decode(all));
    if (!response.ok) throw new ApiError(response.status,result);
    return result as T;
  }
  page(query: Query,cursor: string | null = null,limit = 128): Promise<Page> {return this.call("query-page",{query,cursor,limit});}
  readRevision(key: Uint8Array | string): Promise<RevisionRead> {return this.call("read-revision",{key:hex(key)});}
  writeRevision(revision: string,value: Uint8Array | string | null): Promise<Receipt> {return this.call("write-revision",{revision,value:value===null ? null : hex(value)});}
  subscribe(query: Query): Promise<Session> {return this.call("query",query);}
  resume(token: string): Promise<{id:string;instance:string}> {return this.call("resume",{token});}
  poll(session: Pick<Session,"id"|"instance">,token: string,signal?: AbortSignal): Promise<StreamPage> {return this.call("poll",{id:session.id,instance:session.instance,token},signal);}
  unsubscribe(session: Pick<Session,"id"|"instance">,token: string): Promise<unknown> {return this.call("unsubscribe",{id:session.id,instance:session.instance,token});}
  snapshotAt(position: bigint): Promise<{position:string;rows:Row[]}> {return this.call("snapshot-at",{position:position.toString()});}
  backup(directory: string): Promise<unknown> {return this.call("backup",{directory});}
  verifyBackup(directory: string): Promise<unknown> {return this.call("verify-backup",{directory});}
  createIndex(name: string,prefix: Uint8Array | string,path: string[]): Promise<unknown> {return this.call("index-create",{name,prefix:Array.from(typeof prefix==="string" ? new TextEncoder().encode(prefix) : prefix),path});}
  indexes(): Promise<unknown> {return this.call("index-list",{});}
  searchIndex(name: string,value: Scalar): Promise<{position:string;indexed:boolean;rows:Row[]}> {return this.call("index-search",{name,value});}
  dropIndex(name: string): Promise<unknown> {return this.call("index-drop",{name});}
  compactionAdvice(retainCommits: bigint): Promise<unknown> {return this.call("compaction-advice",{retain_commits:retainCommits.toString()});}
  /** Yield one complete page at a time. Persist rows and token before requesting another.
   * A cancellation after a write is not an abort. No transport retry or implicit resnapshot. */
  async *stream(session: Pick<Session,"id"|"instance">,appliedToken: string,signal?: AbortSignal): AsyncGenerator<StreamPage> {
    let token=appliedToken;
    try {
      while (!signal?.aborted) {
        const page=await this.poll(session,token,signal);
        if (BigInt(page.position)<BigInt(tokenPosition(token))) throw new Error("stream cursor regressed");
        if (page.token!==token) {yield page;token=page.token;}
        else await new Promise<void>(resolve=>{const done=()=>{clearTimeout(timer);signal?.removeEventListener("abort",done);resolve();};const timer=setTimeout(done,25);signal?.addEventListener("abort",done,{once:true});});
      }
    } finally {await this.unsubscribe(session,token).catch(()=>{});}
  }
}
function tokenPosition(token: string): bigint {
  const bytes=unhex(token);
  if (bytes.length<36 || new TextDecoder().decode(bytes.subarray(0,8))!=="ENDBTOK1") throw new Error("invalid local cursor");
  return new DataView(bytes.buffer,bytes.byteOffset,bytes.byteLength).getBigUint64(24,true);
}