import * as l from "@cerulea/lexicon";
import { isBareOrigin, readBytes } from "./http.ts";
import {
  type Did,
  isDid,
  parseAtUri,
  Profile,
  Star,
  starUrl,
} from "./records.ts";

export type Entry = { uri: string; value: unknown };
export type StarEntry = { uri: string; star: Star; link: string };
export type Actor = {
  did: Did;
  handle: string;
  pds: string;
  handleVerified?: boolean;
  avatar?: string;
};

// Public services consulted for accounts Hydrant doesn't track.
const SLINGSHOT = new URL("https://slingshot.microcosm.blue");
const PLC = new URL("https://plc.directory");
const BSKY = new URL("https://public.api.bsky.app");
const CDN = new URL("https://cdn.cerulea.blue");

type Json = Record<string, unknown>;
type Params = Record<string, string | number | boolean | undefined>;
const isJson = (data: unknown): data is Json =>
  typeof data === "object" && data !== null && !Array.isArray(data);

const xrpc = (base: URL, method: string, params: Params): URL => {
  const url = new URL(`/xrpc/${method}`, base);
  for (const [key, value] of Object.entries(params)) {
    if (value !== undefined) url.searchParams.set(key, String(value));
  }
  return url;
};

// A bounded JSON object fetch; 404 is absence rather than failure.
async function getJson(url: URL): Promise<Json | undefined> {
  const res = await fetch(url, {
    headers: { accept: "application/json" },
    redirect: "error",
    signal: AbortSignal.timeout(10000),
  });
  if (res.status === 404) return;
  if (!res.ok || Number(res.headers.get("content-length")) > 4_194_304) {
    throw new Error(`Upstream ${url.origin}: ${res.status}`);
  }
  const bytes = await readBytes(res.body ?? [], 4_194_304);
  const data: unknown = JSON.parse(new TextDecoder().decode(bytes));
  if (!isJson(data)) throw new Error(`Invalid response from ${url.origin}`);
  return data;
}

// For lookups with a fallback, where failure is just absence.
const tryJson = (url: URL) => getJson(url).catch(() => undefined);

const MiniDoc = l.object({
  did: l.string,
  pds: l.string,
  handle: l.optional(l.string),
});

// What we know of an account before checking its avatar.
type Identity = Omit<Actor, "avatar">;

function identity(
  did: Did,
  pds: string,
  claimed: string | undefined,
  handleVerified: boolean,
): Identity | undefined {
  const endpoint = new URL(pds);
  if (!isBareOrigin(endpoint) || endpoint.protocol !== "https:") return;
  const usable = claimed && claimed !== "handle.invalid" &&
    l.handle.validate(claimed).success;
  return {
    did,
    pds: endpoint.href,
    handle: usable ? claimed : did,
    handleVerified,
  };
}

// A DID document's PDS endpoint and the handle it claims.
function plcPds(doc: Json): string | undefined {
  for (const service of Array.isArray(doc.service) ? doc.service : []) {
    if (
      service?.id === "#atproto_pds" &&
      typeof service.serviceEndpoint === "string"
    ) {
      return service.serviceEndpoint;
    }
  }
}
function plcHandle(doc: Json): string | undefined {
  for (const alias of Array.isArray(doc.alsoKnownAs) ? doc.alsoKnownAs : []) {
    const claimed = typeof alias === "string" && alias.startsWith("at://") &&
      alias.slice(5);
    if (claimed && l.handle.validate(claimed).success) return claimed;
  }
}

// Stars whose link we can't render are treated as absent.
function parseStar(entry: Entry): StarEntry | undefined {
  const parsed = Star.validate(entry.value);
  const link = parsed.success ? starUrl(parsed.value.url) : undefined;
  if (parsed.success && link) {
    return { uri: entry.uri, star: parsed.value, link };
  }
}

export class Hydrant {
  constructor(readonly base: URL) {}

  private get(method: string, params: Params): Promise<Json | undefined> {
    return getJson(xrpc(this.base, method, params));
  }

  // Hydrant's identity cache, then the public one.
  private async miniDoc(identifier: string) {
    const method = "blue.microcosm.identity.resolveMiniDoc";
    const doc = await tryJson(xrpc(this.base, method, { identifier })) ??
      await tryJson(xrpc(SLINGSHOT, method, { identifier }));
    const parsed = MiniDoc.validate(doc);
    if (parsed.success && isDid(parsed.value.did)) {
      return { ...parsed.value, did: parsed.value.did };
    }
  }

  // A PLC account whose handle can't be verified still has a claimed one in
  // its DID document, which we show as unverified.
  private async unverified(
    identifier: string,
    pds?: string,
  ): Promise<Identity | undefined> {
    if (!isDid(identifier) || !identifier.startsWith("did:plc:")) return;
    const doc = await getJson(new URL(`/${identifier}`, PLC));
    if (doc?.id !== identifier) return;
    const endpoint = pds ?? plcPds(doc);
    if (!endpoint) return;
    return identity(identifier, endpoint, plcHandle(doc), false);
  }

  async actor(identifier: string): Promise<Actor | undefined> {
    const doc = await this.miniDoc(identifier);
    const known = doc && doc.handle !== "handle.invalid"
      ? identity(doc.did, doc.pds, doc.handle, true)
      : await this.unverified(
        doc?.did ?? identifier,
        doc?.pds,
      );
    if (!known) return;
    const avatar = await this.avatar(known.did);
    return avatar ? { ...known, avatar } : known;
  }

  private async avatar(did: Did): Promise<string | undefined> {
    const profile = await this.get("com.atproto.repo.getRecord", {
      repo: did,
      collection: "app.bsky.actor.profile",
      rkey: "self",
    });
    const parsed = Profile.validate(profile?.value);
    if (parsed.success && parsed.value.avatar) {
      return new URL(`/${did}/${parsed.value.avatar.ref.$link}`, CDN).href;
    }
    // Hydrant only holds tracked profiles; Bluesky's AppView has the rest.
    const view = await tryJson(
      xrpc(BSKY, "app.bsky.actor.getProfile", { actor: did }),
    );
    const url = view?.did === did && typeof view.avatar === "string"
      ? URL.parse(view.avatar)
      : null;
    if (url?.protocol === "https:" && url.hostname === "cdn.bsky.app") {
      return url.href;
    }
  }

  async records(
    did: string,
    collection: string,
    cursor?: string,
    limit = 50,
  ): Promise<{ entries: Entry[]; cursor?: string }> {
    const data = await this.get("com.atproto.repo.listRecords", {
      repo: did,
      collection,
      limit,
      cursor,
      reverse: false,
    });
    if (!data) return { entries: [] };
    const { records, cursor: next } = data;
    if (
      !Array.isArray(records) || records.length > limit ||
      (next !== undefined && typeof next !== "string")
    ) throw new Error("Invalid record page");
    const entries = records.filter((entry): entry is Entry =>
      typeof entry?.uri === "string" &&
      entry.uri.startsWith(`at://${did}/${collection}/`)
    );
    return { entries, cursor: next };
  }

  // Follows a collection's listRecords cursor to the end.
  async *allRecords(did: string, collection: string): AsyncGenerator<Entry> {
    let cursor: string | undefined;
    do {
      const page = await this.records(did, collection, cursor, 100);
      yield* page.entries;
      if (page.cursor && page.cursor === cursor) {
        throw new Error("Repeated Hydrant cursor");
      }
      cursor = page.cursor;
    } while (cursor);
  }

  async record(uri: string): Promise<Entry | undefined> {
    const at = parseAtUri(uri);
    if (!at) return;
    const data = await this.get("com.atproto.repo.getRecord", {
      repo: at.did,
      collection: at.collection,
      rkey: at.rkey,
    });
    if (data?.uri !== uri) return;
    return { uri, value: data.value };
  }

  async star(uri: string): Promise<StarEntry | undefined> {
    const entry = await this.record(uri);
    return entry && parseStar(entry);
  }

  async stars(
    did: string,
    cursor?: string,
  ): Promise<{ stars: StarEntry[]; cursor?: string }> {
    const page = await this.records(did, Star.id, cursor, 25);
    return {
      stars: page.entries.flatMap((entry) => parseStar(entry) ?? []),
      cursor: page.cursor,
    };
  }

  // One page of the repos Hydrant tracks, from its management API.
  private async repoPage(cursor?: string): Promise<string[]> {
    const url = new URL("/repos", this.base);
    url.searchParams.set("limit", "1000");
    if (cursor) url.searchParams.set("cursor", cursor);
    const res = await fetch(url, {
      headers: { accept: "application/json" },
      signal: AbortSignal.timeout(30000),
    });
    if (!res.ok) throw new Error(`Hydrant repos: ${res.status}`);
    const repos: unknown = await res.json();
    if (!Array.isArray(repos) || repos.length > 1000) {
      throw new Error("Invalid repo list");
    }
    const dids: unknown[] = repos.map((row) => row?.did);
    const valid = dids.every((did): did is string => typeof did === "string");
    if (!valid || (cursor && dids.includes(cursor))) {
      throw new Error("Invalid repo page");
    }
    return dids;
  }

  async *repos(): AsyncGenerator<string> {
    let page: string[], cursor: string | undefined;
    do {
      page = await this.repoPage(cursor);
      yield* page;
      cursor = page.at(-1);
    } while (page.length);
  }

  async track(did: string): Promise<void> {
    const res = await fetch(new URL("/repos", this.base), {
      method: "PUT",
      headers: { "content-type": "application/json" },
      body: JSON.stringify([{ did }]),
      signal: AbortSignal.timeout(10000),
    });
    if (!res.ok) throw new Error(`Could not track repo: ${res.status}`);
    await res.body?.cancel();
  }

  // The event stream, resuming from the given event ID.
  stream(cursor: number): URL {
    const url = new URL("/stream", this.base);
    url.protocol = url.protocol === "https:" ? "wss:" : "ws:";
    url.searchParams.set("cursor", String(cursor));
    return url;
  }
}