import chunk from 'lodash/chunk';
import { jsonParse, jsonStringify, UnexpectedError } from 'n8n-workflow';

import { ByteStoreRegistry } from './byte-store-registry';
import { SkippedEntryDeletionError } from './skipped-entry-deletion.error';
import type { ByteStore, JsonEntry, JsonStoreOptions, StorageLocation, Stored } from './types';

const CORRUPT_ENTRY = Symbol('corruptEntry');

const MAX_READ_CONCURRENCY = 50;

/**
 * Stores JSON entries as blobs using pluggable backends.
 * The `version` key is reserved for the schema version: writes reject payloads that contain it.
 */
export class JsonStore<Ref, Payload extends object> {
	private readonly byteStores: ByteStoreRegistry;

	constructor(private readonly options: JsonStoreOptions<Ref>) {
		this.byteStores = new ByteStoreRegistry(options.byteStores);
	}

	registerByteStore(loc: StorageLocation, store: ByteStore) {
		this.byteStores.register(loc, store);
	}

	hasLocation(loc: StorageLocation) {
		return this.byteStores.has(loc);
	}

	async write(ref: Ref, payload: Payload, loc: StorageLocation) {
		const store = this.byteStores.get(loc);
		if ('version' in payload) {
			throw new UnexpectedError('Payload must not contain the reserved `version` key.');
		}
		try {
			const body = Buffer.from(
				jsonStringify({ ...payload, version: this.options.version }),
				'utf-8',
			);
			return await store.write(this.options.key(ref), body, { mimeType: 'application/json' });
		} catch (error) {
			throw this.options.createWriteError(ref, error);
		}
	}

	async read(ref: Ref, loc: StorageLocation) {
		const store = this.byteStores.get(loc);
		const key = this.options.key(ref);
		const bytes = await store.read(key);
		if (!bytes) return null;
		return this.parse(ref, bytes);
	}

	async readMany(refs: Array<Stored<Ref>>) {
		const entries = new Map<string, JsonEntry<Payload>>();
		if (refs.length === 0) return entries;

		await Promise.all(
			[...this.groupByLocation(refs)].map(async ([loc, group]) => {
				const store = this.byteStores.get(loc);
				for (const batch of chunk(group, MAX_READ_CONCURRENCY)) {
					const results = await Promise.all(
						batch.map(async (ref) => await this.tryRead(ref, store)),
					);
					results.forEach((entry, idx) => {
						if (entry) entries.set(this.options.getId(batch[idx]), entry);
					});
				}
			}),
		);

		return entries;
	}

	async delete(refs: Array<Stored<Ref>>) {
		if (refs.length === 0) return;

		await Promise.all(
			[...this.groupByLocation(refs)].map(async ([loc, group]) => {
				const store = this.byteStores.find(loc);
				if (!store) {
					this.options.reportError(new SkippedEntryDeletionError(loc, group.length));
					return;
				}
				await store.delete(group.map((ref) => this.options.key(ref)));
			}),
		);
	}

	// private methods

	private groupByLocation(refs: Array<Stored<Ref>>) {
		const groups = new Map<StorageLocation, Ref[]>();
		for (const ref of refs) {
			const group = groups.get(ref.storedAt) ?? [];
			group.push(ref);
			groups.set(ref.storedAt, group);
		}
		return groups;
	}

	private async tryRead(ref: Ref, store: ByteStore) {
		try {
			const bytes = await store.read(this.options.key(ref));
			if (!bytes) return null;
			return this.parse(ref, bytes);
		} catch (error) {
			if (this.isCorruptEntryError(error)) {
				this.options.reportError(error);
				return null;
			}
			throw error;
		}
	}

	private parse(ref: Ref, bytes: Buffer) {
		try {
			const entry = jsonParse<JsonEntry<Payload>>(bytes.toString('utf-8'));
			if (entry.version !== this.options.version) {
				throw new Error(`Unsupported entry version: ${String(entry.version)}`);
			}
			return entry;
		} catch (error) {
			throw this.markCorrupt(this.options.createCorruptedError(ref, error));
		}
	}

	private markCorrupt<E extends Error>(error: E): E {
		Object.defineProperty(error, CORRUPT_ENTRY, { value: true });
		return error;
	}

	private isCorruptEntryError(error: unknown): error is Error {
		return error instanceof Error && CORRUPT_ENTRY in error;
	}
}