<script lang="ts">
	import { Alert, Button } from '$lib/components/common'
	import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
	import Drawer from '$lib/components/common/drawer/Drawer.svelte'
	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
	import Path from '$lib/components/Path.svelte'
	import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte'
	import { KafkaTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen'
	import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
	import { getTriggerWorkspace } from '$lib/components/triggers/triggerWorkspace'
	import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
	import { withForkConflictRetry } from '$lib/utils/forkConflict'
	import Section from '$lib/components/Section.svelte'
	import { Loader2, RotateCcw } from 'lucide-svelte'
	import Label from '$lib/components/Label.svelte'
	import KafkaTriggersConfigSection from './KafkaTriggersConfigSection.svelte'
	import { untrack, type Snippet } from 'svelte'
	import TriggerEditorToolbar from '../TriggerEditorToolbar.svelte'
	import PermissionedAsLine from '../PermissionedAsLine.svelte'
	import { saveKafkaTriggerFromCfg } from './utils'
	import { getHandlerType, handleConfigChange, type Trigger } from '../utils'
	import Tabs from '$lib/components/common/tabs/Tabs.svelte'
	import Tab from '$lib/components/common/tabs/Tab.svelte'
	import TriggerRetriesAndErrorHandler from '../TriggerRetriesAndErrorHandler.svelte'
	import TriggerAdvancedBadges from '../TriggerAdvancedBadges.svelte'
	import { deepEqual } from 'fast-equals'
	import { useTriggerDraftSync } from '../useTriggerDraftSync.svelte'
	import LocalDraftBanner from '$lib/components/LocalDraftBanner.svelte'
	import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte'
	import TriggerSuspendedJobsModal from '../TriggerSuspendedJobsModal.svelte'
	import TriggerFilters from '../TriggerFilters.svelte'
	import Select from '$lib/components/select/Select.svelte'
	import Toggle from '$lib/components/Toggle.svelte'

	interface Props {
		useDrawer?: boolean
		description?: Snippet | undefined
		hideTarget?: boolean
		hideTooltips?: boolean
		isEditor?: boolean
		allowDraft?: boolean
		trigger?: Trigger
		customLabel?: Snippet
		isDeployed?: boolean
		cloudDisabled?: boolean
		onConfigChange?: (cfg: Record<string, any>, saveDisabled: boolean, updated: boolean) => void
		onCaptureConfigChange?: (cfg: Record<string, any>, isValid: boolean) => void
		onUpdate?: (path?: string) => void
		onDelete?: () => void
		onReset?: () => void
	}

	let {
		useDrawer = true,
		description = undefined,
		hideTarget = false,
		hideTooltips = false,
		isEditor = false,
		allowDraft = false,
		trigger = undefined,
		customLabel,
		isDeployed = false,
		cloudDisabled = false,
		onConfigChange = undefined,
		onCaptureConfigChange = undefined,
		onUpdate = undefined,
		onDelete = undefined,
		onReset = undefined
	}: Props = $props()
	const triggerWs = getTriggerWorkspace()
	const wsId = $derived(triggerWs?.() ?? $workspaceStore)

	let drawer: Drawer | undefined = $state()
	let is_flow: boolean = $state(false)
	let initialPath = $state('')
	let edit = $state(true)
	let itemKind: 'flow' | 'script' = $state('script')
	let script_path = $state('')
	let initialScriptPath = $state('')
	let fixedScriptPath = $state('')
	let path: string = $state('')
	let pathError = $state('')
	let mode = $state<TriggerMode>('enabled')
	let dirtyPath = $state(false)
	let can_write = $state(true)
	let drawerLoading = $state(true)
	let showLoading = $state(false)
	let initialConfig: Record<string, any> | undefined = undefined
	let extra_perms = $state<Record<string, any> | undefined>(undefined)
	let kafkaCfgValid = $state(false)
	let kafkaResourcePath = $state('')
	let kafkaCfg: Record<string, any> = $state({})
	let autoOffsetReset = $state('latest')
	let autoCommit = $state(true)
	let deploymentLoading = $state(false)
	let permissionedAs = $state<string | undefined>(undefined)
	let selectedPermissionedAs = $state<string | undefined>(undefined)
	let preservePermissionedAs = $state(false)
	let resetLoading = $state(false)
	let optionTabSelected: 'error_handler' | 'retries' = $state('error_handler')
	let errorHandlerSelected: ErrorHandler = $state('slack')
	let error_handler_path: string | undefined = $state()
	let error_handler_args: Record<string, any> = $state({})
	let retry: Retry | undefined = $state()
	let filters: { key: string; value: any }[] = $state([])
	let filterLogic = $state<'and' | 'or'>('and')

	let suspendedJobsModal = $state<TriggerSuspendedJobsModal | null>(null)
	let originalConfig = $state<Record<string, any> | undefined>(undefined)
	let resetConfirmOpen = $state(false)

	const isValid = $derived(
		!!kafkaResourcePath &&
			kafkaCfgValid &&
			kafkaCfg.topics &&
			kafkaCfg.topics.length > 0 &&
			kafkaCfg.topics.every((b) => /^[a-zA-Z0-9-_.]+$/.test(b))
	)

	let hasChanged = $derived(!deepEqual(getSaveCfg(), originalConfig ?? {}))
	const saveDisabled = $derived(
		pathError !== '' ||
			!isValid ||
			drawerLoading ||
			!can_write ||
			emptyString(script_path) ||
			emptyString(kafkaResourcePath) ||
			kafkaCfg.topics.length === 0 ||
			kafkaCfg.topics.some((t) => emptyString(t)) ||
			!hasChanged
	)
	const kafkaConfig = $derived.by(getSaveCfg)

	const draftSync = useTriggerDraftSync({
		itemKind: 'trigger_kafka',
		path: () => initialPath,
		workspace: () => wsId,
		drawerLoading: () => drawerLoading,
		getCfg: () => kafkaConfig,
		applyCfg: loadTriggerConfig,
		deployed: () => originalConfig
	})
	const captureConfig = $derived.by(untrack(() => isEditor) ? getCaptureConfig : () => ({}))

	$effect(() => {
		is_flow = itemKind === 'flow'
	})

	export async function openEdit(
		ePath: string,
		isFlow: boolean,
		defaultConfig?: Record<string, any>,
		fixedScriptPath_?: string
	) {
		let loadingTimeout = setTimeout(() => {
			showLoading = true
		}, 100) // Do not show loading spinner for the first 100ms
		drawerLoading = true
		try {
			drawer?.openDrawer()
			initialPath = ePath
			itemKind = isFlow ? 'flow' : 'script'
			edit = true
			dirtyPath = false
			fixedScriptPath = fixedScriptPath_ ?? ''
			const { overlay: draftOverlay, noDeployed } = await loadTrigger(defaultConfig)
			// Draft-only triggers open as "new trigger prefilled from the
			// draft" — no deployed row exists, so saving must CREATE (the
			// update endpoint 404s).
			edit = !noDeployed
			originalConfig = structuredClone($state.snapshot(getSaveCfg()))
			if (draftOverlay) loadTriggerConfig(draftOverlay)
			if (!defaultConfig) {
				initialConfig = structuredClone($state.snapshot(getSaveCfg()))
			}
			await draftSync.maybeRestore()
		} catch (err) {
			sendUserToast(`Could not load Kafka trigger: ${err}`, true)
		} finally {
			clearTimeout(loadingTimeout)
			drawerLoading = false
			showLoading = false
		}
	}

	export async function openNew(
		nis_flow: boolean,
		fixedScriptPath_?: string,
		nDefaultValues?: Record<string, any>
	) {
		let loadingTimeout = setTimeout(() => {
			showLoading = true
		}, 100) // Do not show loading spinner for the first 100ms
		drawerLoading = true
		try {
			drawer?.openDrawer()
			is_flow = nis_flow
			edit = false
			itemKind = nis_flow ? 'flow' : 'script'
			kafkaResourcePath = nDefaultValues?.kafka_resource_path ?? ''
			kafkaCfg = {
				group_id: nDefaultValues?.group_id ?? '',
				topics: nDefaultValues?.topics ?? ['']
			}
			autoOffsetReset = nDefaultValues?.auto_offset_reset ?? 'latest'
			autoCommit = nDefaultValues?.auto_commit ?? true
			initialScriptPath = ''
			fixedScriptPath = fixedScriptPath_ ?? ''
			script_path = fixedScriptPath
			path = nDefaultValues?.path ?? ''
			initialPath = ''
			dirtyPath = false
			error_handler_path = nDefaultValues?.error_handler_path ?? undefined
			error_handler_args = nDefaultValues?.error_handler_args ?? {}
			retry = nDefaultValues?.retry ?? undefined
			filters = nDefaultValues?.filters ?? []
			filterLogic = 'and'
			errorHandlerSelected = getHandlerType(error_handler_path ?? '')
			mode = nDefaultValues?.mode ?? 'enabled'
			permissionedAs = undefined
			selectedPermissionedAs = undefined
			preservePermissionedAs = false
			originalConfig = undefined
		} finally {
			clearTimeout(loadingTimeout)
			drawerLoading = false
			showLoading = false
		}
	}

	function loadTriggerConfig(cfg?: Record<string, any>): void {
		script_path = cfg?.script_path
		initialScriptPath = cfg?.script_path
		is_flow = cfg?.is_flow
		path = cfg?.path
		kafkaResourcePath = cfg?.kafka_resource_path
		kafkaCfg = {
			group_id: cfg?.group_id,
			topics: cfg?.topics
		}
		autoOffsetReset = cfg?.auto_offset_reset ?? 'latest'
		autoCommit = cfg?.auto_commit ?? true
		mode = cfg?.mode ?? 'enabled'
		extra_perms = cfg?.extra_perms
		can_write = canWrite(path, cfg?.extra_perms, $userStore)
		error_handler_path = cfg?.error_handler_path
		error_handler_args = cfg?.error_handler_args ?? {}
		retry = cfg?.retry
		filters = cfg?.filters ?? []
		filterLogic = cfg?.filter_logic ?? 'and'
		errorHandlerSelected = getHandlerType(error_handler_path ?? '')
		permissionedAs = cfg?.permissioned_as
		selectedPermissionedAs = cfg?.permissioned_as
		preservePermissionedAs = !!cfg?.permissioned_as
	}

	/** See `NatsTriggerEditorInner.loadTrigger` for the rationale. */
	async function loadTrigger(
		defaultConfig?: Record<string, any>
	): Promise<{ overlay: Record<string, any> | undefined; noDeployed: boolean }> {
		if (defaultConfig) {
			loadTriggerConfig(defaultConfig)
			return { overlay: undefined, noDeployed: false }
		}
		const s = await KafkaTriggerService.getKafkaTrigger({
			workspace: wsId!,
			path: initialPath,
			getDraft: true
		})
		const { draft: draftFromBackend, ...deployedTrigger } = (s ?? {}) as any
		loadTriggerConfig(deployedTrigger)
		return {
			noDeployed: !!(s as any)?.no_deployed,
			overlay: draftFromBackend
				? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
				: undefined
		}
	}

	function getSaveCfg(): Record<string, any> {
		return {
			path,
			script_path,
			is_flow,
			kafka_resource_path: kafkaResourcePath,
			group_id: kafkaCfg.group_id,
			topics: kafkaCfg.topics,
			filters,
			filter_logic: filterLogic,
			auto_offset_reset: autoOffsetReset,
			auto_commit: autoCommit,
			mode,
			extra_perms: extra_perms,
			error_handler_path,
			error_handler_args,
			retry,
			permissioned_as: selectedPermissionedAs,
			preserve_permissioned_as: preservePermissionedAs || undefined
		}
	}

	async function updateTrigger(): Promise<void> {
		deploymentLoading = true
		const previousPath = initialPath
		const cfg = getSaveCfg()
		const isSaved = await saveKafkaTriggerFromCfg(
			initialPath,
			cfg,
			edit,
			wsId!,
			usedTriggerKinds
		)
		if (isSaved) {
			draftSync.discard(previousPath, getSaveCfg())
			onUpdate?.(cfg.path)
			originalConfig = structuredClone($state.snapshot(getSaveCfg()))
			initialPath = cfg.path
			initialScriptPath = cfg.script_path
			if (mode !== 'suspended') {
				drawer?.closeDrawer()
			}
		}
		deploymentLoading = false
	}

	function getCaptureConfig(): Record<string, any> {
		return {
			kafka_resource_path: kafkaResourcePath,
			group_id: kafkaCfg.group_id,
			topics: structuredClone($state.snapshot(kafkaCfg.topics)),
			path
		}
	}

	async function resetOffsets() {
		resetLoading = true
		try {
			await KafkaTriggerService.resetKafkaOffsets({
				workspace: wsId!,
				path: initialPath
			})
			sendUserToast(
				'Offset reset triggered. The consumer will restart and re-read from the beginning.'
			)
		} catch (error) {
			sendUserToast(error.body || error.message, true)
		} finally {
			resetLoading = false
			resetConfirmOpen = false
		}
	}

	async function handleToggleMode(newMode: TriggerMode) {
		const previousMode = mode
		mode = newMode
		if (!trigger?.draftConfig) {
			const ok = await withForkConflictRetry(
				(force) =>
					KafkaTriggerService.setKafkaTriggerMode({
						path: initialPath,
						workspace: wsId ?? '',
						requestBody: { mode: newMode, force }
					}),
				'Kafka trigger'
			)
			if (!ok) {
				mode = previousMode
				return
			}
			sendUserToast(`${capitalize(newMode)} Kafka trigger ${initialPath}`)
			onUpdate?.(initialPath)
		}
		if (originalConfig) {
			originalConfig['mode'] = newMode
		}
	}

	$effect(() => {
		const args = [captureConfig, isValid] as const
		untrack(() => onCaptureConfigChange?.(...args))
	})

	$effect(() => {
		if (!drawerLoading) {
			handleConfigChange(kafkaConfig, initialConfig, saveDisabled, edit, onConfigChange)
		}
	})
</script>

<ConfirmationModal
	title="Reset consumer offset"
	confirmationText="Reset"
	open={resetConfirmOpen}
	loading={resetLoading}
	onConfirmed={resetOffsets}
	onCanceled={() => (resetConfirmOpen = false)}
>
	This will re-process all messages from the beginning of the topic. The consumer will restart
	automatically.
</ConfirmationModal>

{#if mode === 'suspended'}
	<TriggerSuspendedJobsModal
		bind:this={suspendedJobsModal}
		triggerPath={path}
		triggerKind="kafka"
		{hasChanged}
		onToggleMode={handleToggleMode}
		runnableConfig={{
			path: script_path,
			kind: itemKind,
			retry,
			errorHandlerPath: error_handler_path,
			errorHandlerArgs: error_handler_args
		}}
	/>
{/if}

{#if useDrawer}
	<Drawer size="800px" bind:this={drawer}>
		<DrawerContent
			bannerReserved={draftSync.hasBaseline}
			title={edit
				? can_write
					? `Edit Kafka trigger ${initialPath}`
					: `Kafka trigger ${initialPath}`
				: 'New Kafka trigger'}
			on:close={drawer.closeDrawer}
		>
			{#snippet actions()}
				{@render actionsButtons('sm')}
			{/snippet}
			{#snippet banner()}
				<LocalDraftBanner
					show={draftSync.hasDraft}
					getDeployed={() => draftSync.deployed}
					reserveSpace={draftSync.hasBaseline}
					getCurrent={() => draftSync.current}
					onDiscard={() => draftSync.resetToDeployed(initialPath)}
					disabled={!can_write}
				/>
			{/snippet}
			{@render config()}
		</DrawerContent>
	</Drawer>
{:else}
	<Section label={!customLabel ? 'Kafka trigger' : ''} headerClass="grow min-w-0 h-[30px]">
		{#snippet header()}
			{#if customLabel}
				{@render customLabel()}
			{/if}
		{/snippet}
		{#snippet action()}
			{@render actionsButtons('xs')}
		{/snippet}
		{@render config()}
	</Section>
{/if}

{#snippet actionsButtons(size: 'xs' | 'sm' = 'sm')}
	{#if !drawerLoading}
		<TriggerEditorToolbar
			{trigger}
			permissions={drawerLoading || !can_write ? 'none' : 'create'}
			{mode}
			{allowDraft}
			{edit}
			isLoading={deploymentLoading}
			{isDeployed}
			{saveDisabled}
			onUpdate={updateTrigger}
			{onReset}
			{onDelete}
			onToggleMode={handleToggleMode}
			{cloudDisabled}
			{suspendedJobsModal}
		/>
	{/if}
{/snippet}

{#snippet config()}
	{#if drawerLoading}
		{#if showLoading}
			<Loader2 class="animate-spin" />
		{/if}
	{:else}
		<PermissionedAsLine
			{permissionedAs}
			{path}
			onPermissionedAsChange={(pa, preserve) => {
				selectedPermissionedAs = pa
				preservePermissionedAs = preserve
			}}
		/>
		<div class="flex flex-col gap-4">
			{#if description}
				{@render description()}
			{/if}
			{#if !hideTooltips}
				<Alert title="Info" type="info" size="xs">
					{#if edit}
						Changes can take up to 30 seconds to take effect.
					{:else}
						Kafka consumers can take up to 30 seconds to start.
					{/if}
				</Alert>
			{/if}
		</div>
		<div class="flex flex-col gap-12 mt-6">
			{#if mode === 'suspended'}
				<TriggerSuspendedJobsAlert {suspendedJobsModal} />
			{/if}
			<div class="flex flex-col gap-4">
				<Label label="Path">
					<Path
						workspaceOverride={wsId}
						bind:dirty={dirtyPath}
						bind:error={pathError}
						bind:path
						{initialPath}
						checkInitialPathExistence={!edit}
						namePlaceholder="kafka_trigger"
						kind="kafka_trigger"
						disabled={!can_write}
						disableEditing={!can_write}
					/>
				</Label>
			</div>

			{#if !hideTarget}
				<Section label="Runnable">
					<TriggerRunnablePicker
						workspace={wsId}
						{fixedScriptPath}
						bind:itemKind
						bind:scriptPath={script_path}
						{initialScriptPath}
						canWrite={can_write}
						isOperator={!!$userStore?.operator}
					>
						{#snippet createButton()}
							{#if emptyString(script_path)}
								<Button
									btnClasses="ml-4"
									variant="accent"
									size="xs"
									disabled={!can_write}
									href={itemKind === 'flow' ? '/flows/add?hub=65' : '/scripts/add?hub=hub%2F19659'}
									target="_blank">Create from template</Button
								>
							{/if}
						{/snippet}
					</TriggerRunnablePicker>
				</Section>
			{/if}

			<KafkaTriggersConfigSection
				bind:kafkaCfgValid
				bind:kafkaResourcePath
				bind:kafkaCfg
				{path}
				{can_write}
				showTestingBadge={isEditor}
			/>

			<Section label="Advanced" collapsable>
				{#snippet header()}
					<TriggerAdvancedBadges
						{error_handler_path}
						{retry}
						extraBadges={[
							{ name: 'Earliest offset', active: autoOffsetReset !== 'latest' },
							{ name: 'Manual commit', active: !autoCommit },
							{ name: 'Filters', active: filters.length > 0 }
						]}
					/>
				{/snippet}
				<div class="flex flex-col gap-6">
					<Label label="Initial offset">
						{#snippet header()}
							<span class="text-2xs text-tertiary ml-2">
								Only applies when no committed offset exists
							</span>
						{/snippet}
						<Select
							items={[
								{ label: 'Latest (new messages only)', value: 'latest' },
								{ label: 'Earliest (from beginning)', value: 'earliest' }
							]}
							bind:value={autoOffsetReset}
							disabled={!can_write}
						/>
					</Label>

					<div class="flex flex-col gap-2">
						<Label label="Auto-commit offsets">
							{#snippet header()}
								<span class="text-2xs text-tertiary ml-2">
									Automatically commit offsets after receiving each message
								</span>
							{/snippet}
							<Toggle bind:checked={autoCommit} disabled={!can_write} />
						</Label>
						{#if !autoCommit}
							<Alert title="Manual commit mode" type="info" size="xs">
								Offsets will not be committed automatically. Use <code
									>wmill.commit_kafka_offsets(trigger_path, topic, partition, offset)</code
								>
								in Python or
								<code>wmill.commitKafkaOffsets(triggerPath, topic, partition, offset)</code> in TypeScript
								with the values from the event payload. The consumer collects all pending commits and
								commits the highest offset for each topic/partition pair.
							</Alert>
						{/if}
					</div>

					{#if edit && can_write}
						<Label label="Consumer offset">
							{#snippet header()}
								<span class="text-2xs text-tertiary ml-2">
									Force re-read all messages from the beginning
								</span>
							{/snippet}
							<Button
								variant="default"
								size="xs"
								startIcon={{ icon: RotateCcw }}
								disabled={resetLoading}
								loading={resetLoading}
								onclick={() => (resetConfirmOpen = true)}
							>
								Reset offset to earliest
							</Button>
						</Label>
					{/if}

					<TriggerFilters
						bind:filters
						bind:filterLogic
						disabled={!can_write}
						payloadBase64Encoded
					/>

					<div class="min-h-96">
						<Tabs bind:selected={optionTabSelected}>
							<Tab value="error_handler" label="Error Handler" />
							<Tab value="retries" label="Retries" />
						</Tabs>
						<div class="mt-4">
							<TriggerRetriesAndErrorHandler
								workspace={wsId}
								{optionTabSelected}
								{itemKind}
								{can_write}
								bind:errorHandlerSelected
								bind:error_handler_path
								bind:error_handler_args
								bind:retry
							/>
						</div>
					</div>
				</div>
			</Section>
			<div class="pb-8" />
		</div>
	{/if}
{/snippet}
