import type {
	PutObjectCommandInput,
	DeleteObjectsCommandInput,
	ListObjectsV2CommandInput,
	S3ClientConfig,
} from '@aws-sdk/client-s3';
import {
	S3Client,
	HeadBucketCommand,
	PutObjectCommand,
	GetObjectCommand,
	HeadObjectCommand,
	DeleteObjectCommand,
	DeleteObjectsCommand,
	ListObjectsV2Command,
} from '@aws-sdk/client-s3';
import { Logger } from '@n8n/backend-common';
import { streamToBuffer } from '@n8n/backend-network';
import { Service } from '@n8n/di';
import { ensureError } from '@n8n/utils/errors/ensure-error';
import chunk from 'lodash/chunk';
import { UnexpectedError } from 'n8n-workflow';
import { createHash } from 'node:crypto';
import { PassThrough, Readable, pipeline } from 'node:stream';

import { ObjectStoreConfig } from './object-store.config';
import type { MetadataResponseHeaders } from './types';
import { createFixedSizeChunker } from '../stream-utils';
import type { PreWriteBlobMetadata } from '../types';

/** How many per-key delete failures to name in the error before truncating, to keep the message bounded. */
const MAX_REPORTED_DELETE_ERRORS = 5;

@Service()
export class ObjectStoreService {
	private s3Client: S3Client;

	private isReady = false;

	private bucket: string;

	constructor(
		private readonly logger: Logger,
		private readonly s3Config: ObjectStoreConfig,
	) {
		const { bucket } = s3Config;
		if (bucket.name === '') {
			throw new UnexpectedError(
				'External storage bucket name not configured. Please set `N8N_EXTERNAL_STORAGE_S3_BUCKET_NAME`.',
			);
		}

		this.bucket = bucket.name;
		this.s3Client = new S3Client(this.getClientConfig());
	}

	/** This generates the config for the S3Client to make it work in all various auth configurations */
	getClientConfig() {
		const { bucket, credentials, maxAttempts } = this.s3Config;
		const clientConfig: S3ClientConfig = {};
		const { endpoint } = this;
		if (endpoint) {
			clientConfig.endpoint = endpoint;
			clientConfig.forcePathStyle = this.s3Config.forcePathStyle;
		}
		if (bucket.region.length) {
			clientConfig.region = bucket.region;
		}
		if (!credentials.authAutoDetect) {
			clientConfig.credentials = {
				accessKeyId: credentials.accessKey,
				secretAccessKey: credentials.accessSecret,
			};
		}
		clientConfig.maxAttempts = maxAttempts;
		return clientConfig;
	}

	async init() {
		const { bucket, forcePathStyle } = this.s3Config;
		this.logger.info(
			`S3 binary storage configured: endpoint=${this.endpoint ?? 'default'}, bucket=${this.bucket}, region=${bucket.region || 'none'}, forcePathStyle=${forcePathStyle}`,
		);
		await this.checkConnection();
		this.setReady(true);
	}

	setReady(newState: boolean) {
		this.isReady = newState;
	}

	/**
	 * Confirm that the configured bucket exists and the caller has permission to access it.
	 */
	async checkConnection() {
		if (this.isReady) return;

		const { endpoint } = this;

		try {
			this.logger.debug('Checking connection to S3 bucket', { bucket: this.bucket, endpoint });
			const command = new HeadBucketCommand({ Bucket: this.bucket });
			const { startupTimeoutMs } = this.s3Config;
			await this.s3Client.send(
				command,
				// NOTE: startupTimeoutMs of 0 means no timeout.
				startupTimeoutMs > 0 ? { abortSignal: AbortSignal.timeout(startupTimeoutMs) } : {},
			);
		} catch (e) {
			const error = ensureError(e);
			const message = `Failed to connect to S3 at ${endpoint ?? 'the default S3 endpoint'} (bucket: ${this.bucket}): ${error.message}. Check that N8N_EXTERNAL_STORAGE_S3_HOST includes the port, that N8N_EXTERNAL_STORAGE_S3_PROTOCOL matches the endpoint, and that the endpoint is reachable.`;
			this.logger.error(message);
			throw new UnexpectedError(message, { cause: error });
		}
	}

	/**
	 * Upload an object to the configured bucket.
	 */
	async put(filename: string, buffer: Buffer, metadata: PreWriteBlobMetadata = {}) {
		try {
			const params: PutObjectCommandInput = {
				Bucket: this.bucket,
				Key: filename,
				Body: buffer,
				ContentLength: buffer.length,
				ContentMD5: createHash('md5').update(buffer).digest('base64'),
			};

			if (metadata.fileName) {
				params.Metadata = { filename: encodeURIComponent(metadata.fileName) };
			}

			if (metadata.mimeType) {
				params.ContentType = metadata.mimeType;
			}

			const { Body: _body, ...logParams } = params;
			this.logger.debug('Sending PUT request to S3', { params: logParams });
			const command = new PutObjectCommand(params);
			return await this.s3Client.send(command);
		} catch (e) {
			this.handleS3Error(e);
		}
	}

	/**
	 * Download an object as a stream or buffer from the configured bucket.
	 *
	 * In `stream` mode, pass `chunkSize` to guarantee that the returned stream
	 * emits chunks of exactly that many bytes (the final chunk may be smaller).
	 * Without it, chunk boundaries follow whatever the underlying socket emits,
	 * which can break consumers like S3 multipart upload that treat each emitted chunk as a fixed-size unit.
	 */
	async get(fileId: string, opts: { mode: 'buffer' }): Promise<Buffer>;
	async get(fileId: string, opts: { mode: 'stream'; chunkSize?: number }): Promise<Readable>;
	async get(
		fileId: string,
		{ mode, chunkSize = 0 }: { mode: 'stream' | 'buffer'; chunkSize?: number },
	): Promise<Buffer | Readable> {
		this.logger.debug('Sending GET request to S3', { bucket: this.bucket, key: fileId });

		const command = new GetObjectCommand({
			Bucket: this.bucket,
			Key: fileId,
		});

		try {
			if (mode === 'stream') {
				const abortController = new AbortController();
				const { Body: body } = await this.s3Client.send(command, {
					abortSignal: abortController.signal,
				});
				if (!body) throw new UnexpectedError('Received empty response body');

				if (!(body instanceof Readable)) {
					throw new UnexpectedError('Expected stream but received different type', {
						extra: { bodyType: typeof body },
					});
				}

				// Wrap to prevent socket pool exhaustion when callers destroy the
				// stream early. AbortController lets the SDK free the socket slot
				// properly. See: https://github.com/aws/aws-sdk-js-v3/issues/6691
				const wrapper = new PassThrough();
				let bodyFullyConsumed = false;

				body.on('end', () => {
					bodyFullyConsumed = true;
				});

				wrapper.on('close', () => {
					if (!bodyFullyConsumed) {
						abortController.abort();
						body.destroy();
					}
				});

				body.on('error', (error) => wrapper.destroy(error));
				body.pipe(wrapper);

				if (chunkSize > 0) {
					const rechunker = createFixedSizeChunker(chunkSize);
					pipeline(wrapper, rechunker, () => {}); // Error/destroy propagation is handled via stream events on `rechunker`
					return rechunker;
				}
				return wrapper;
			}

			const { Body: body } = await this.s3Client.send(command);
			if (!body) throw new UnexpectedError('Received empty response body');

			return await streamToBuffer(body as Readable);
		} catch (e) {
			if (e instanceof UnexpectedError) throw e;
			this.handleS3Error(e);
		}
	}

	/**
	 * Retrieve metadata for an object in the configured bucket.
	 */
	async getMetadata(fileId: string): Promise<MetadataResponseHeaders> {
		try {
			const command = new HeadObjectCommand({
				Bucket: this.bucket,
				Key: fileId,
			});

			this.logger.debug('Sending HEAD request to S3', { bucket: this.bucket, key: fileId });
			const response = await this.s3Client.send(command);

			// Convert response to the expected format for backward compatibility
			const headers: MetadataResponseHeaders = {};

			if (response.ContentType) headers['content-type'] = response.ContentType;
			if (response.ContentLength) headers['content-length'] = String(response.ContentLength);
			if (response.ETag) headers.etag = response.ETag;
			if (response.LastModified) headers['last-modified'] = response.LastModified.toUTCString();

			// Add metadata with the expected prefix format
			if (response.Metadata) {
				Object.entries(response.Metadata).forEach(([key, value]) => {
					headers[`x-amz-meta-${key.toLowerCase()}`] =
						key === 'filename' ? decodeURIComponent(value) : value;
				});
			}

			return headers;
		} catch (e) {
			this.handleS3Error(e);
		}
	}

	/**
	 * Delete a single object in the configured bucket.
	 */
	async deleteOne(fileId: string) {
		try {
			const command = new DeleteObjectCommand({
				Bucket: this.bucket,
				Key: fileId,
			});

			this.logger.debug('Sending DELETE request to S3', { bucket: this.bucket, key: fileId });
			return await this.s3Client.send(command);
		} catch (e) {
			this.handleS3Error(e);
		}
	}

	/**
	 * Delete objects with a common prefix in the configured bucket.
	 */
	async deleteMany(prefix: string) {
		const objects = await this.list(prefix);

		await this.deleteByKeys(objects.map(({ key }) => key));
	}

	/**
	 * Delete objects by exact key in the configured bucket, in batches of up
	 * to 1000 keys, the `DeleteObjects` limit.
	 */
	async deleteByKeys(keys: string[]) {
		if (keys.length === 0) return;

		try {
			for (const batch of chunk(keys, 1000)) {
				const params: DeleteObjectsCommandInput = {
					Bucket: this.bucket,
					Delete: {
						Objects: batch.map((key) => ({ Key: key })),
					},
				};

				this.logger.debug('Sending DELETE MANY request to S3', {
					bucket: this.bucket,
					objectCount: batch.length,
				});

				// `DeleteObjects` reports per-key failures in the response rather than failing the request
				const { Errors: errors } = await this.s3Client.send(new DeleteObjectsCommand(params));

				if (errors && errors.length > 0) {
					this.logger.error('Failed to delete objects from S3', {
						bucket: this.bucket,
						failures: errors.map((e) => ({ key: e.Key, code: e.Code, message: e.Message })),
					});

					const summary = errors
						.slice(0, MAX_REPORTED_DELETE_ERRORS)
						.map((e) => `${e.Key ?? '<unknown key>'} (${e.Code ?? '?'}: ${e.Message ?? '?'})`)
						.join(', ');
					throw new UnexpectedError(
						`Failed to delete ${errors.length} of ${batch.length} objects: ${summary}`,
					);
				}
			}
		} catch (e) {
			this.handleS3Error(e);
		}
	}

	/**
	 * List objects with a common prefix in the configured bucket.
	 */
	async list(prefix: string) {
		const items = [];
		let isTruncated = true;
		let continuationToken;

		try {
			while (isTruncated) {
				const listPage = await this.getListPage(prefix, continuationToken);

				if (listPage.contents?.length > 0) {
					items.push(...listPage.contents);
				}

				isTruncated = listPage.isTruncated;
				continuationToken = listPage.nextContinuationToken;
			}

			return items;
		} catch (e) {
			this.handleS3Error(e);
		}
	}

	/**
	 * Fetch a page of objects with a common prefix in the configured bucket.
	 */
	async getListPage(prefix: string, continuationToken?: string) {
		try {
			const params: ListObjectsV2CommandInput = {
				Bucket: this.bucket,
				Prefix: prefix,
			};

			if (continuationToken) {
				params.ContinuationToken = continuationToken;
			}

			this.logger.debug('Sending list request to S3', { bucket: this.bucket, prefix });
			const command = new ListObjectsV2Command(params);
			const response = await this.s3Client.send(command);

			// Convert response to match expected format for compatibility
			const contents =
				response.Contents?.map((item) => ({
					key: item.Key ?? '',
					lastModified: item.LastModified?.toISOString() ?? '',
					eTag: item.ETag ?? '',
					size: item.Size ?? 0,
					storageClass: item.StorageClass ?? '',
				})) ?? [];

			return {
				contents,
				isTruncated: response.IsTruncated ?? false,
				nextContinuationToken: response.NextContinuationToken,
			};
		} catch (e) {
			this.handleS3Error(e);
		}
	}

	/** Custom S3 endpoint, or `undefined` to let the SDK resolve the default. */
	private get endpoint() {
		const { host, protocol } = this.s3Config;
		return host ? `${protocol}://${host}` : undefined;
	}

	private handleS3Error(e: unknown): never {
		const error = ensureError(e);
		throw new UnexpectedError(`Request to S3 failed: ${error.message}`, { cause: error });
	}
}