/* eslint-disable svelte/prefer-svelte-reactivity */
import { err, expect, map, ok, type OkType, type Result } from '$lib/result';
import {
	ComAtprotoIdentityResolveHandle,
	ComAtprotoRepoGetRecord,
	ComAtprotoRepoListRecords
} from '@atcute/atproto';
import { Client as AtcuteClient, simpleFetchHandler } from '@atcute/client';
import { safeParse, type Blob as AtpBlob, type Handle, type InferOutput } from '@atcute/lexicons';
import {
	isDid,
	parseResourceUri,
	type ActorIdentifier,
	type AtprotoDid,
	type Cid,
	type Did,
	type Nsid,
	type RecordKey,
	type ResourceUri
} from '@atcute/lexicons/syntax';
import type {
	InferInput,
	InferXRPCBodyOutput,
	ObjectSchema,
	RecordKeySchema,
	RecordSchema,
	XRPCQueryMetadata
} from '@atcute/lexicons/validations';
import * as v from '@atcute/lexicons/validations';
import { MiniDocQuery, type MiniDoc } from './slingshot';
import { BacklinksQuery, type Backlinks, type BacklinksSource } from './constellation';
import type { Records, XRPCProcedures } from '@atcute/lexicons/ambient';
import { cache as rawCache, ttl } from '$lib/cache';
import { AppBskyActorProfile } from '@atcute/bluesky';
import { WebSocket } from '@soffinal/websocket';
import type { Notification } from './stardust';
import type { OAuthUserAgent } from '@atcute/oauth-browser-client';
import { timestampFromCursor, toCanonicalUri, toResourceUri } from '$lib';
import { constellationUrl, httpToDidWeb, slingshotUrl, spacedustUrl } from '.';

export type RecordOutput<Output> = { uri: ResourceUri; cid: Cid | undefined; record: Output };

const cacheWithHandles = rawCache.define(
	'resolveHandle',
	async (handle: Handle): Promise<AtprotoDid> => {
		const res = await fetchMicrocosm(slingshotUrl, ComAtprotoIdentityResolveHandle.mainSchema, {
			handle
		});
		if (!res.ok) throw new Error(res.error);
		return res.value.did as AtprotoDid;
	}
);

const cacheWithDidDocs = cacheWithHandles.define(
	'resolveDidDoc',
	async (identifier: ActorIdentifier): Promise<MiniDoc> => {
		const res = await fetchMicrocosm(slingshotUrl, MiniDocQuery, {
			identifier
		});
		if (!res.ok) throw new Error(res.error);
		return res.value;
	}
);

const cacheWithRecords = cacheWithDidDocs.define('fetchRecord', async (uri: ResourceUri) => {
	const parsedUri = parseResourceUri(uri);
	if (!parsedUri.ok) throw new Error(`can't parse resource uri: ${parsedUri.error}`);
	const { repo, collection, rkey } = parsedUri.value;
	const res = await fetchMicrocosm(slingshotUrl, ComAtprotoRepoGetRecord.mainSchema, {
		repo,
		collection: collection!,
		rkey: rkey!
	});
	if (!res.ok) throw new Error(res.error);
	return res.value;
});

const cache = cacheWithRecords;

export const invalidateRecordCache = async (uri: ResourceUri) => {
	console.log(`invalidating cached for ${uri}`);
	await cache.invalidate('fetchRecord', `fetchRecord~${uri}`);
};
export const setRecordCache = (uri: ResourceUri, record: unknown) =>
	cache.set('fetchRecord', `fetchRecord~${uri}`, record, ttl);

export const xhrPost = (
	url: string,
	body: Blob | File,
	headers: Record<string, string> = {},
	onProgress?: (uploaded: number, total: number) => void
	// eslint-disable-next-line @typescript-eslint/no-explicit-any
): Promise<Result<any, { error: string; message: string }>> => {
	return new Promise((resolve) => {
		const xhr = new XMLHttpRequest();
		xhr.open('POST', url);

		if (onProgress && xhr.upload)
			xhr.upload.onprogress = (event: ProgressEvent) => {
				if (event.lengthComputable) onProgress(event.loaded, event.total);
			};

		Object.keys(headers).forEach((key) => xhr.setRequestHeader(key, headers[key]));

		xhr.onload = () => {
			if (xhr.status >= 200 && xhr.status < 300) resolve(ok(JSON.parse(xhr.responseText)));
			else resolve(err(JSON.parse(xhr.responseText)));
		};

		xhr.onerror = () => resolve(err({ error: 'xhr_error', message: 'network error' }));
		xhr.onabort = () => resolve(err({ error: 'xhr_error', message: 'upload aborted' }));
		xhr.send(body);
	});
};

export type UploadStatus =
	| { stage: 'auth' }
	| { stage: 'uploading'; progress?: number }
	| { stage: 'processing'; progress?: number }
	| { stage: 'complete' };

export type Auth = {
	atcute: AtcuteClient;
} & MiniDoc;

export class AtpClient {
	public user: Auth | null = $state(null);

	async login(agent: OAuthUserAgent): Promise<Result<null, string>> {
		try {
			const rpc = new AtcuteClient({ handler: agent });
			const res = await rpc.get('com.atproto.server.getSession');
			if (!res.ok) throw res.data.error;
			this.user = {
				atcute: rpc,
				did: res.data.did,
				handle: res.data.handle,
				pds: agent.session.info.aud as `${string}:${string}`,
				signing_key: ''
			};
		} catch (error) {
			return err(`failed to login: ${error}`);
		}

		return ok(null);
	}

	async getRecordUri<
		Collection extends Nsid,
		TObject extends ObjectSchema & { shape: { $type: v.LiteralSchema<Collection> } },
		TKey extends RecordKeySchema,
		Schema extends RecordSchema<TObject, TKey>,
		Output extends InferInput<Schema>
	>(
		schema: Schema,
		uri: ResourceUri,
		noCache?: boolean
	): Promise<Result<RecordOutput<Output>, string>> {
		const parsedUri = expect(parseResourceUri(uri));
		if (parsedUri.collection !== schema.object.shape.$type.expected)
			return err(
				`collections don't match: ${parsedUri.collection} != ${schema.object.shape.$type.expected}`
			);
		return await this.getRecord(schema, parsedUri.repo!, parsedUri.rkey!, noCache);
	}

	async getRecord<
		Collection extends Nsid,
		TObject extends ObjectSchema & { shape: { $type: v.LiteralSchema<Collection> } },
		TKey extends RecordKeySchema,
		Schema extends RecordSchema<TObject, TKey>,
		Output extends InferInput<Schema>
	>(
		schema: Schema,
		repo: ActorIdentifier,
		rkey: RecordKey,
		noCache?: boolean
	): Promise<Result<RecordOutput<Output>, string>> {
		const collection = schema.object.shape.$type.expected;

		try {
			const uri = toResourceUri({ repo, collection, rkey, fragment: undefined });
			if (noCache) await invalidateRecordCache(uri);
			const rawValue = await cache.fetchRecord(uri);

			const parsed = safeParse(schema, rawValue.value);
			if (!parsed.ok) return err(parsed.message);

			return ok({
				uri: rawValue.uri,
				cid: rawValue.cid,
				record: parsed.value as Output
			});
		} catch (e) {
			return err(String(e));
		}
	}

	async getProfile(
		repo?: ActorIdentifier,
		noCache?: boolean
	): Promise<Result<AppBskyActorProfile.Main, string>> {
		repo = repo ?? this.user?.did;
		if (!repo) return err('not authenticated');
		return map(
			await this.getRecord(AppBskyActorProfile.mainSchema, repo, 'self', noCache),
			(d) => d.record
		);
	}

	async listRecords<Collection extends keyof Records>(
		ident: ActorIdentifier,
		collection: Collection,
		cursor?: string,
		limit: number = 100
	): Promise<
		Result<InferXRPCBodyOutput<(typeof ComAtprotoRepoListRecords.mainSchema)['output']>, string>
	> {
		const auth = this.user;
		if (!auth) return err('not authenticated');
		const docRes = await resolveDidDoc(ident);
		if (!docRes.ok) return docRes;
		const atp =
			auth.did === docRes.value.did
				? auth.atcute
				: new AtcuteClient({ handler: simpleFetchHandler({ service: docRes.value.pds }) });
		const res = await atp.get('com.atproto.repo.listRecords', {
			params: {
				repo: docRes.value.did,
				collection,
				cursor,
				limit,
				reverse: false
			}
		});
		if (!res.ok) return err(`${res.data.error}: ${res.data.message ?? 'no details'}`);

		for (const record of res.data.records) setRecordCache(record.uri, record);

		return ok(res.data);
	}

	async listRecordsUntil<Collection extends keyof Records>(
		ident: ActorIdentifier,
		collection: Collection,
		cursor?: string,
		timestamp: number = -1,
		limit: number = 100
	): Promise<ReturnType<typeof this.listRecords>> {
		const data: OkType<Awaited<ReturnType<typeof this.listRecords>>> = {
			records: [],
			cursor
		};

		let end = false;
		while (!end) {
			const res = await this.listRecords(ident, collection, data.cursor, limit);
			if (!res.ok) return res;
			data.cursor = res.value.cursor;
			data.records.push(...res.value.records);
			end = data.records.length === 0 || !data.cursor;
			if (!end && timestamp > 0) {
				const cursorTimestamp = timestampFromCursor(data.cursor);
				if (cursorTimestamp === undefined) {
					console.warn(
						'could not parse timestamp from cursor, stopping fetch to prevent infinite loop:',
						data.cursor
					);
					end = true;
				} else if (cursorTimestamp <= timestamp) {
					end = true;
				} else {
					console.info(
						`${ident}: continuing to fetch ${collection}, on ${cursorTimestamp} until ${timestamp}`
					);
				}
			}
		}

		return ok(data);
	}

	async getBacklinks(
		subject: ResourceUri,
		source: BacklinksSource,
		filterBy?: Did[],
		limit?: number
	): Promise<Result<Backlinks, string>> {
		const { repo, collection, rkey } = expect(parseResourceUri(subject));
		const did = await resolveHandle(repo);
		if (!did.ok) return err(`cant resolve handle: ${did.error}`);

		const timeout = new Promise<null>((resolve) => setTimeout(() => resolve(null), 2000));
		const query = fetchMicrocosm(constellationUrl, BacklinksQuery, {
			subject: collection ? toCanonicalUri({ did: did.value, collection, rkey: rkey! }) : did.value,
			source,
			limit: limit || 100,
			did: filterBy
		});

		const results = await Promise.race([query, timeout]);
		if (!results) return err('cant fetch backlinks: timeout');

		return results;
	}

	async getServiceAuth(lxm: keyof XRPCProcedures, exp: number): Promise<Result<string, string>> {
		const auth = this.user;
		if (!auth) return err('not authenticated');
		const serviceAuthUrl = new URL(`${auth.pds}xrpc/com.atproto.server.getServiceAuth`);
		serviceAuthUrl.searchParams.append('aud', httpToDidWeb(auth.pds));
		serviceAuthUrl.searchParams.append('lxm', 'com.atproto.repo.uploadBlob');
		serviceAuthUrl.searchParams.append('exp', exp.toString()); // 30 minutes

		const serviceAuthResponse = await auth.atcute.handler(
			`${serviceAuthUrl.pathname}${serviceAuthUrl.search}`,
			{
				method: 'GET'
			}
		);
		if (!serviceAuthResponse.ok) {
			const error = await serviceAuthResponse.text();
			return err(`failed to get service auth: ${error}`);
		}
		const serviceAuth = await serviceAuthResponse.json();
		return ok(serviceAuth.token);
	}

	async uploadBlob(
		blob: Blob,
		onProgress?: (progress: number) => void
	): Promise<Result<AtpBlob<string>, string>> {
		const auth = this.user;
		if (!auth) return err('not authenticated');
		const tokenResult = await this.getServiceAuth(
			'com.atproto.repo.uploadBlob',
			Math.floor(Date.now() / 1000) + 60
		);
		if (!tokenResult.ok) return tokenResult;
		const result = await xhrPost(
			`${auth.pds}xrpc/com.atproto.repo.uploadBlob`,
			blob,
			{ authorization: `Bearer ${tokenResult.value}` },
			(uploaded, total) => onProgress?.(uploaded / total)
		);
		if (!result.ok) return err(`upload failed: ${result.error.message}`);
		return ok(result.value.blob);
	}

	async uploadVideo(
		blob: Blob,
		mimeType: string,
		onStatus?: (status: UploadStatus) => void
	): Promise<Result<AtpBlob<string>, string>> {
		const auth = this.user;
		if (!auth) return err('not authenticated');

		onStatus?.({ stage: 'auth' });
		const tokenResult = await this.getServiceAuth(
			'com.atproto.repo.uploadBlob',
			Math.floor(Date.now() / 1000) + 60 * 30
		);
		if (!tokenResult.ok) return tokenResult;

		onStatus?.({ stage: 'uploading' });
		const uploadUrl = new URL('https://video.bsky.app/xrpc/app.bsky.video.uploadVideo');
		uploadUrl.searchParams.append('did', auth.did);
		uploadUrl.searchParams.append('name', 'video');

		const uploadResult = await xhrPost(
			uploadUrl.toString(),
			blob,
			{
				Authorization: `Bearer ${tokenResult.value}`,
				'Content-Type': mimeType
			},
			(uploaded, total) => onStatus?.({ stage: 'uploading', progress: uploaded / total })
		);
		if (!uploadResult.ok) return err(`failed to upload video: ${uploadResult.error}`);
		const jobStatus = uploadResult.value;
		let videoBlobRef: AtpBlob<string> = jobStatus.blob;

		onStatus?.({ stage: 'processing' });
		while (!videoBlobRef) {
			await new Promise((resolve) => setTimeout(resolve, 1000));

			const statusResponse = await fetch(
				`https://video.bsky.app/xrpc/app.bsky.video.getJobStatus?jobId=${jobStatus.jobId}`
			);

			if (!statusResponse.ok) {
				const error = await statusResponse.json();
				// reuse blob
				if (error.error === 'already_exists' && error.blob) {
					videoBlobRef = error.blob;
					break;
				}
				return err(`failed to get job status: ${error.message || error.error}`);
			}

			const status = await statusResponse.json();
			if (status.jobStatus.blob) {
				videoBlobRef = status.jobStatus.blob;
			} else if (status.jobStatus.state === 'JOB_STATE_FAILED') {
				return err(`video processing failed: ${status.jobStatus.error || 'unknown error'}`);
			} else if (status.jobStatus.progress !== undefined) {
				onStatus?.({
					stage: 'processing',
					progress: status.jobStatus.progress / 100
				});
			}
		}

		onStatus?.({ stage: 'complete' });
		return ok(videoBlobRef);
	}
}

// export const newPublicClient = async (ident: ActorIdentifier) => {
// 	const atp = new AtpClient();
// 	const didDoc = await resolveDidDoc(ident);
// 	if (!didDoc.ok) {
// 		console.error('failed to resolve did doc', didDoc.error);
// 		return atp;
// 	}
// 	atp.atcute = new AtcuteClient({ handler: simpleFetchHandler({ service: didDoc.value.pds }) });
// 	atp.user = didDoc.value;
// 	return atp;
// };

export const resolveHandle = (identifier: ActorIdentifier) => {
	if (isDid(identifier)) return Promise.resolve(ok(identifier as AtprotoDid));

	return cache
		.resolveHandle(identifier)
		.then((did) => ok(did))
		.catch((e) => err(String(e)));
};

export const resolveDidDoc = (ident: ActorIdentifier) =>
	cache
		.resolveDidDoc(ident)
		.then((doc) => ok(doc))
		.catch((e) => err(String(e)));

type NotificationsStreamEncoder = WebSocket.Encoder<undefined, Notification>;
export type NotificationsStream = WebSocket<NotificationsStreamEncoder>;
export type NotificationsStreamEvent = WebSocket.Event<NotificationsStreamEncoder>;

export const streamNotifications = (
	subjects: Did[],
	...sources: BacklinksSource[]
): NotificationsStream => {
	const url = new URL(spacedustUrl);
	url.protocol = 'wss:';
	url.pathname = '/subscribe';
	const searchParams = [];
	sources.every((source) => searchParams.push(['wantedSources', source]));
	subjects.every((subject) => searchParams.push(['wantedSubjectDids', subject]));
	subjects.every((subject) => searchParams.push(['wantedSubjects', `at://${subject}`]));
	searchParams.push(['instant', 'true']);
	url.search = `?${new URLSearchParams(searchParams)}`;
	// console.log(`streaming notifications: ${url}`);
	const encoder = WebSocket.getDefaultEncoder<undefined, Notification>();
	const ws = new WebSocket<typeof encoder>(url.toString(), {
		encoder
	});
	return ws;
};

const fetchMicrocosm = async <
	Schema extends XRPCQueryMetadata,
	Input extends Schema['params'] extends ObjectSchema ? InferOutput<Schema['params']> : undefined,
	Output extends InferXRPCBodyOutput<Schema['output']>
>(
	api: URL,
	schema: Schema,
	params: Input,
	init?: RequestInit
): Promise<Result<Output, string>> => {
	if (!schema.output || schema.output.type === 'blob') return err('schema must be blob');
	api.pathname = `/xrpc/${schema.nsid}`;
	api.search = params
		? `?${new URLSearchParams(Object.entries(params).flatMap(([k, v]) => (v === undefined ? [] : [[k, String(v)]])))}`
		: '';
	try {
		const body = await fetchJson(api, init);
		if (!body.ok) return err(body.error);
		const parsed = safeParse(schema.output.schema, body.value);
		if (!parsed.ok) return err(parsed.message);
		return ok(parsed.value as Output);
	} catch (error) {
		return err(`FetchError: ${error}`);
	}
};

const fetchJson = async (url: URL, init?: RequestInit): Promise<Result<unknown, string>> => {
	try {
		const response = await fetch(url, init);
		const body = await response.json();
		if (response.status === 400) return err(`${body.error}: ${body.message}`);
		if (response.status !== 200) return err(`UnexpectedStatusCode: ${response.status}`);
		return ok(body);
	} catch (error) {
		return err(`FetchError: ${error}`);
	}
};