import { describe, expect, it } from 'vitest'
import type { ScriptLang } from '$lib/gen'
import {
	autoOutputAsset,
	generatePipelineDraft,
	type PipelineOutputKind
} from './pipelineTemplates'

// The seeded draft asset (`autoOutputAsset`, stored as `outputAssets` and used
// by resolveGraph for inactive-draft node identity) must match the asset
// identity the deploy-time / wasm parser infers from the generated body. The
// parser keeps the suffix after `s3://` verbatim (see backend
// `parse_asset_syntax`), so a default-storage object's path carries a leading
// slash (`s3:///key` → `/key`). If the seed and the body's write URI disagree,
// the preview renders a duplicate node and a phantom post-deploy drift. This
// pins the two in lockstep so that class of drift can't regress.

// Mirror of the parser's S3 path extraction for a raw `s3://…` URI.
function canonicalS3Key(uri: string): string {
	return uri.replace(/^s3:\/\//, '')
}

const S3_KINDS: PipelineOutputKind[] = ['s3_parquet', 's3_object']
const LANGS: ScriptLang[] = ['bun', 'python3', 'duckdb']

describe('pipelineTemplates S3 seed/body parity', () => {
	for (const language of LANGS) {
		for (const outputKind of S3_KINDS) {
			it(`${language} ${outputKind}: seeded asset path matches the body's S3 write URI`, () => {
				const output = autoOutputAsset(outputKind, 'demo', language)
				expect(output).toBeDefined()
				const asset = output!

				// The seed must carry the default-storage leading slash so it
				// matches the identity the parser infers from the generated body.
				expect(asset.kind).toBe('s3object')
				expect(asset.path.startsWith('/')).toBe(true)

				const body = generatePipelineDraft({
					language,
					outputKind,
					output: asset,
					triggers: []
				})

				// Every S3 URI the generated body emits must canonicalize back to
				// the seeded asset path — the write target especially.
				const uris = body.match(/s3:\/\/[^'"`)\s]+/g) ?? []
				expect(uris.length).toBeGreaterThan(0)
				for (const uri of uris) {
					// Runtime form must stay triple-slash (default storage); a bare
					// `s3://key` would target a named storage `key` at run time.
					expect(uri.startsWith('s3:///')).toBe(true)
					expect(canonicalS3Key(uri)).toBe(asset.path)
				}
			})
		}
	}
})

// `{partition}` substitutes to the partition IDENTITY string (e.g. `2026-07-05T23`,
// `2026-W27`, `2026-07`), which is NOT a valid DuckDB TIMESTAMP literal for any
// sub-day / non-daily grain — so a naive `WHERE ts = TIMESTAMP {partition}`
// raises a Conversion Error for hourly/weekly/monthly. The runtime injects a
// `wm_partition(ts)` macro (format from the same source that stamps the
// identity), so the scaffold teaches the one grain-agnostic filter line.
describe('pipelineTemplates materialize partition filter', () => {
	const materializeBody = () =>
		generatePipelineDraft({
			language: 'duckdb',
			outputKind: 'materialize',
			output: autoOutputAsset('materialize', 'demo', 'duckdb'),
			input: { kind: 'ducklake', path: 'main/orders' },
			triggers: []
		})

	it('teaches the grain-agnostic wm_partition filter', () => {
		const body = materializeBody()
		expect(body).toContain(`WHERE wm_partition(<ts_col>) = {partition}`)
	})

	it('never scaffolds the naive `TIMESTAMP {partition}` cast, nor a raw strftime format', () => {
		const body = materializeBody()
		// The footgun cast must never appear (comment or SQL) now that the macro
		// hides the format entirely.
		expect(body).not.toMatch(/TIMESTAMP\s*\{partition\}/)
		// No hand-written strftime format to drift from the resolver.
		expect(body).not.toContain('strftime')
	})
})

describe('pipelineTemplates: auto-derived reads drop the redundant // on', () => {
	const draft = (
		input: { kind: 'ducklake' | 's3object' | 'datatable'; path: string } | undefined,
		triggers: Parameters<typeof generatePipelineDraft>[0]['triggers'],
		language: ScriptLang = 'duckdb',
		outputKind: PipelineOutputKind = 'materialize'
	) =>
		generatePipelineDraft({
			language,
			outputKind,
			output: autoOutputAsset(outputKind, 'demo', language),
			input,
			triggers
		})

	it('omits // on for a ducklake input the body reads (auto-derived)', () => {
		const body = draft({ kind: 'ducklake', path: 'main/orders' }, [
			{ kind: 'asset', ref: 'ducklake://main/orders' }
		])
		expect(body).not.toContain('on ducklake://main/orders')
		// The body still attaches and reads the table (`ducklake://main` →
		// `lake.orders`), which is what wires the cascade now that the explicit
		// `// on` is gone.
		expect(body).toContain(`ATTACH 'ducklake://main'`)
		expect(body).toContain('lake.orders')
	})

	it('omits // on for an s3 input the body reads', () => {
		const body = draft({ kind: 's3object', path: 'raw/events.parquet' }, [
			{ kind: 'asset', ref: 's3://raw/events.parquet' }
		])
		expect(body).not.toContain('on s3://raw/events.parquet')
	})

	it('keeps // on for a datatable input (not auto-derived)', () => {
		const body = draft({ kind: 'datatable', path: 'main/dt' }, [
			{ kind: 'asset', ref: 'datatable://main/dt' }
		])
		expect(body).toContain('on datatable://main/dt')
	})

	it('keeps // on for a native trigger', () => {
		const body = draft(undefined, [{ kind: 'schedule', path: undefined }])
		expect(body).toContain('on schedule')
	})

	it('keeps // on when the body does not read the input (postgres, bash)', () => {
		// postgres/bash bodies ignore `input`, so dropping `// on` would leave the
		// script with neither an explicit trigger nor an inferred read → no cascade.
		const pg = draft(
			{ kind: 'ducklake', path: 'main/orders' },
			[{ kind: 'asset', ref: 'ducklake://main/orders' }],
			'postgresql',
			'datatable'
		)
		expect(pg).toContain('on ducklake://main/orders')

		const bash = draft(
			{ kind: 's3object', path: 'raw/events.parquet' },
			[{ kind: 'asset', ref: 's3://raw/events.parquet' }],
			'bash',
			's3_object'
		)
		expect(bash).toContain('on s3://raw/events.parquet')
	})
})
