import { randomUUID } from "node:crypto";
import { ObjectId } from "mongodb";
import { authCondition } from "$lib/server/auth";
import { collections } from "$lib/server/database";
import {
	releaseWorkspaceRetirements,
	retiredWorkspaceFilter,
	retireAgentWorkspaces,
	WorkspaceRetirementConflict,
} from "$lib/server/agent/retireWorkspaces";
import type { Conversation } from "$lib/types/Conversation";
import type { Generation } from "$lib/types/Generation";
import { GENERATION_REGISTRATION_STALE_MS } from "$lib/server/generation/writer";

export class ConversationDeletionConflict extends Error {
	constructor(message = "Conversation changed while it was being deleted") {
		super(message);
		this.name = "ConversationDeletionConflict";
	}
}

const ACTIVE_GENERATION_DELETE_MESSAGE =
	"Stop the active generation before deleting this conversation";
const CONVERSATION_DELETION_LEASE_MS = 15 * 60 * 1_000;
const STALE_GENERATION_REGISTRATION_MESSAGE =
	"Generation registration expired before the run started";

function isDuplicateKeyError(cause: unknown): boolean {
	return (
		typeof cause === "object" &&
		cause !== null &&
		"code" in cause &&
		(cause as { code?: number }).code === 11_000
	);
}

async function assertNoRunningGeneration(conversationId: Conversation["_id"]): Promise<void> {
	const running = await collections.generations.findOne(
		{ conversationId, status: "running" },
		{ projection: { _id: 1 } }
	);
	if (running) throw new ConversationDeletionConflict(ACTIVE_GENERATION_DELETE_MESSAGE);
}

async function reconcileStaleGenerationRegistration(
	conversation: Conversation,
	owner: Record<string, unknown>,
	now: Date
): Promise<Conversation> {
	const registration = conversation.generationRegistration;
	if (!registration) return conversation;
	const staleBefore = new Date(now.getTime() - GENERATION_REGISTRATION_STALE_MS);
	if (registration.claimedAt > staleBefore) {
		throw new ConversationDeletionConflict(ACTIVE_GENERATION_DELETE_MESSAGE);
	}

	// Consume the globally unique generation id before releasing a crashed
	// registration. If the delayed producer wins this insert race, its running
	// row is observed below and deletion remains blocked. If this tombstone wins,
	// that producer can never create a running orphan after the conversation is
	// gone.
	try {
		await collections.generations.insertOne({
			_id: new ObjectId(),
			generationId: registration.generationId,
			conversationId: conversation._id,
			messageId: registration.messageId,
			...(conversation.userId
				? { userId: conversation.userId }
				: conversation.sessionId
					? { sessionId: conversation.sessionId }
					: {}),
			status: "error",
			seq: 0,
			lastHeartbeatAt: now,
			startedAt: registration.claimedAt,
			endedAt: now,
			error: STALE_GENERATION_REGISTRATION_MESSAGE,
			...(conversation.mode ? { mode: conversation.mode } : {}),
			...(conversation.workspaceId ? { workspaceId: conversation.workspaceId } : {}),
			...(conversation.mode === "agent"
				? { agent: { sourceSequence: 0, controlsOpen: false } }
				: {}),
			createdAt: now,
			updatedAt: now,
		} satisfies Generation);
	} catch (cause) {
		if (!isDuplicateKeyError(cause)) throw cause;
	}

	const generation = await collections.generations.findOne(
		{ generationId: registration.generationId },
		{ projection: { conversationId: 1, status: 1 } }
	);
	if (generation?.conversationId.equals(conversation._id) && generation.status === "running") {
		throw new ConversationDeletionConflict(ACTIVE_GENERATION_DELETE_MESSAGE);
	}

	const cleared = await collections.conversations.findOneAndUpdate(
		{
			_id: conversation._id,
			...owner,
			updatedAt: conversation.updatedAt,
			...(conversation.mode === undefined
				? { mode: { $exists: false } }
				: { mode: conversation.mode }),
			...(conversation.workspaceId
				? { workspaceId: conversation.workspaceId }
				: { workspaceId: { $exists: false } }),
			deletion: { $exists: false },
			"generationRegistration.id": registration.id,
		},
		{ $unset: { generationRegistration: "" } },
		{ returnDocument: "after", includeResultMetadata: false }
	);
	if (!cleared) throw new ConversationDeletionConflict();
	return cleared;
}

/**
 * Delete one conversation without allowing a resumed Job to outlive its
 * workspace record. Every linked workspace stays retirement-leased through
 * the conversation CAS and its own exact-incarnation delete.
 */
export async function deleteConversationWithAgentWorkspaces(
	conversation: Conversation,
	locals: App.Locals
): Promise<void> {
	const owner = authCondition(locals);
	conversation = await reconcileStaleGenerationRegistration(conversation, owner, new Date());
	// A successful delete may never strand a detached Generation after its
	// conversation/workspace authority disappears. Existing runs fail closed;
	// callers can stop them and retry once their terminal write has settled.
	await assertNoRunningGeneration(conversation._id);
	// Registration and deletion contend on this conversation write before any
	// linked workspace is retired. A fresh registration therefore cannot leave a
	// live conversation pointing at a Job which this failed delete already killed.
	const deletionId = randomUUID();
	const deletionClaimedAt = new Date();
	const deletionStaleBefore = new Date(
		deletionClaimedAt.getTime() - CONVERSATION_DELETION_LEASE_MS
	);
	const claimedConversation = await collections.conversations.findOneAndUpdate(
		{
			_id: conversation._id,
			...owner,
			updatedAt: conversation.updatedAt,
			...(conversation.mode === undefined
				? { mode: { $exists: false } }
				: { mode: conversation.mode }),
			...(conversation.workspaceId
				? { workspaceId: conversation.workspaceId }
				: { workspaceId: { $exists: false } }),
			generationRegistration: { $exists: false },
			$or: [
				{ deletion: { $exists: false } },
				{ "deletion.claimedAt": { $lt: deletionStaleBefore } },
			],
		},
		{ $set: { deletion: { id: deletionId, claimedAt: deletionClaimedAt } } },
		{ returnDocument: "after", includeResultMetadata: false }
	);
	if (!claimedConversation) {
		throw new ConversationDeletionConflict();
	}
	const releaseConversationDeletion = () =>
		collections.conversations.updateOne(
			{ _id: conversation._id, "deletion.id": deletionId },
			{ $unset: { deletion: "" } }
		);
	// Close the check -> claim race before any destructive workspace work. A
	// Generation whose registration inserted and released just before this claim
	// is now visible; once this claim exists, no new production registration can
	// pass its `deletion: absent` predicate.
	try {
		await assertNoRunningGeneration(conversation._id);
	} catch (cause) {
		await releaseConversationDeletion();
		throw cause;
	}

	let linkedWorkspaces;
	try {
		linkedWorkspaces = await collections.workspaces
			.find({
				...owner,
				$or: [
					{ conversationId: conversation._id },
					...(conversation.workspaceId ? [{ _id: conversation.workspaceId }] : []),
				],
			})
			.toArray();
	} catch (cause) {
		await releaseConversationDeletion();
		throw cause;
	}
	let retired;
	try {
		retired = await retireAgentWorkspaces(linkedWorkspaces, locals);
	} catch (cause) {
		await releaseConversationDeletion();
		if (cause instanceof WorkspaceRetirementConflict) {
			throw new ConversationDeletionConflict();
		}
		throw cause;
	}

	// Defense in depth for non-production/manual writers which do not participate
	// in the conversation registration protocol.
	try {
		await assertNoRunningGeneration(conversation._id);
	} catch (cause) {
		await Promise.all([
			releaseWorkspaceRetirements(retired, locals),
			releaseConversationDeletion(),
		]);
		throw cause;
	}

	let deleted;
	try {
		deleted = await collections.conversations.deleteOne({
			_id: conversation._id,
			"deletion.id": deletionId,
		});
	} catch (cause) {
		await Promise.all([
			releaseWorkspaceRetirements(retired, locals),
			releaseConversationDeletion(),
		]);
		throw cause;
	}
	if (deleted.deletedCount !== 1) {
		await Promise.all([
			releaseWorkspaceRetirements(retired, locals),
			releaseConversationDeletion(),
		]);
		throw new ConversationDeletionConflict();
	}

	for (const workspace of retired) {
		const removed = await collections.workspaces.deleteOne(
			retiredWorkspaceFilter(workspace, locals)
		);
		if (removed.deletedCount !== 1) {
			// The owning conversation is already gone, but the lease still blocks
			// provisioning and the Job was confirmed stopped. Keep the tombstone for
			// an operator-visible retry instead of broadening this delete.
			throw new ConversationDeletionConflict();
		}
	}
}