// ATProto client surface for safekeeper.
//
// Uses the official @atproto packages:
//   - @atproto/identity    -> IdResolver handles handle ↔ DID resolution and
//                             DID-doc → PDS endpoint extraction (plc + did:web).
//   - @atproto/api         -> AtpAgent gives us typed, lexicon-generated calls
//                             for every `com.atproto.*` endpoint.
//
// Streaming concession: AtpAgent's `getRepo` / `getBlob` buffer the whole
// response into a `Uint8Array`. For backup workloads where accounts can be
// GB-scale, we fall back to raw `fetch` for those two endpoints so the body
// can be piped straight into storage.

import { AtpAgent } from '@atproto/api';
import { IdResolver, MemoryCache } from '@atproto/identity';
import { UpstreamError } from '../../error';

// Shared resolver, cached for 6 hours. All PDS/handle lookups go through it.
const didCache = new MemoryCache(6 * 60 * 60 * 1000);
export const idResolver = new IdResolver({ didCache });

// Small pool of agents keyed by PDS endpoint so we don't rebuild an XrpcClient
// for every call.
const agentPool = new Map<string, AtpAgent>();
function agent(pds: string): AtpAgent {
	const key = normalisePds(pds);
	let a = agentPool.get(key);
	if (!a) {
		a = new AtpAgent({ service: key });
		agentPool.set(key, a);
	}
	return a;
}

function normalisePds(pds: string): string {
	return pds.endsWith('/') ? pds.slice(0, -1) : pds;
}

// --- Identity ---

// Resolves a DID to its current atproto identity (handle + pds + signingKey).
// Throws if the DID can't be resolved or has no atproto PDS service entry.
export async function resolveAtprotoData(did: string) {
	try {
		return await idResolver.did.resolveAtprotoData(did);
	} catch (err) {
		throw new UpstreamError(
			`could not resolve DID ${did}: ${(err as Error).message}`
		);
	}
}

// Resolve a handle to a DID (DNS TXT or /.well-known fallback).
export async function resolveHandleToDid(handle: string): Promise<string> {
	const did = await idResolver.handle.resolve(handle);
	if (!did) throw new UpstreamError(`could not resolve handle ${handle}`);
	return did;
}

// --- Sync endpoints (typed via AtpAgent) ---

export async function* iterateRepos(
	pds: string,
	limit = 1000
): AsyncGenerator<{
	did: string;
	head: string;
	rev: string;
	active?: boolean;
	status?: string;
}> {
	const a = agent(pds);
	let cursor: string | undefined;
	do {
		const res = await a.com.atproto.sync.listRepos({ limit, cursor });
		for (const r of res.data.repos) yield r;
		cursor = res.data.cursor;
	} while (cursor);
}

export async function getRepoStatus(pds: string, did: string) {
	const res = await agent(pds).com.atproto.sync.getRepoStatus({ did });
	return res.data;
}

export async function* iterateBlobs(
	pds: string,
	did: string,
	since?: string,
	limit = 1000
): AsyncGenerator<string> {
	const a = agent(pds);
	let cursor: string | undefined;
	do {
		const res = await a.com.atproto.sync.listBlobs({
			did,
			since,
			limit,
			cursor
		});
		for (const c of res.data.cids) yield c;
		cursor = res.data.cursor;
	} while (cursor);
}

export async function describeRepo(pds: string, repo: string) {
	const res = await agent(pds).com.atproto.repo.describeRepo({ repo });
	return res.data;
}

// --- Streaming blob/repo fetches ---

// Returns the raw Response so callers can pipe `.body` straight into storage.
// Throws on non-2xx; surfaces `Retry-After` seconds when available.
export async function streamRepo(
	pds: string,
	did: string,
	since?: string
): Promise<Response> {
	const url = new URL('/xrpc/com.atproto.sync.getRepo', normalisePds(pds));
	url.searchParams.set('did', did);
	if (since) url.searchParams.set('since', since);
	return expectOk(await fetch(url), url);
}

export async function streamBlob(
	pds: string,
	did: string,
	cid: string
): Promise<Response> {
	const url = new URL('/xrpc/com.atproto.sync.getBlob', normalisePds(pds));
	url.searchParams.set('did', did);
	url.searchParams.set('cid', cid);
	return expectOk(await fetch(url), url);
}

export class StreamXrpcError extends Error {
	constructor(
		public readonly endpoint: string,
		public readonly status: number,
		public readonly body: string,
		public readonly retryAfterSeconds: number | null
	) {
		super(`XRPC ${endpoint} -> ${status}: ${body.slice(0, 200)}`);
	}
}

async function expectOk(res: Response, url: URL): Promise<Response> {
	if (res.ok) return res;
	const body = await res.text().catch(() => '');
	const ra = res.headers.get('retry-after');
	throw new StreamXrpcError(
		url.pathname,
		res.status,
		body,
		ra ? Number.parseInt(ra, 10) || null : null
	);
}