// Typed Releases Queries
//
// A release, for these queries, is a commit the moment it starts serving
// traffic: the service-overview rollups pre-extract `vcs.ref.head.revision` as
// `CommitSha` and key on it, so every row here is a GROUP BY over the same
// splice the services list already reads. Nothing scans the raw traces table.

import { type DateTime, Schema } from "effect"
import * as T from "@maple-dev/effect-orm/clickhouse"
import * as CH from "@maple-dev/effect-orm/expr"
import {
	param,
	from,
	fromQuery,
	type CHQuery,
	type CompiledQueryRowSchema,
} from "@maple-dev/effect-orm/clickhouse"
import type { ColumnDefs } from "@maple-dev/effect-orm/clickhouse"
import { ErrorEventsByTime, ServiceOverviewSpans, orgIdParam, utcSecondsParam } from "../tables"
import { CHNumber } from "../schema"
import { DateTimeUtcFromWarehouse } from "../../datetime"
import { serviceOverviewWhereConditions } from "./query-helpers"
import { serviceOverviewWindows, serviceWindowTiersForBucket } from "./services"

/**
 * At most this many (service, environment, commit) rows per request. A fleet
 * that deploys every push accumulates thousands of shas in a month; the page
 * lists the newest and says how many it left out.
 */
export const RELEASES_LIST_CAP = 500

/**
 * Values an SDK writes into `vcs.ref.head.revision` when it has no revision
 * to report. They are not releases: the services table drops them from its
 * deploy cell, and one of them showing up here as "errors from 0" was the
 * first thing the page did against live data.
 */
export const PLACEHOLDER_COMMIT_SHAS: readonly string[] = ["", "unknown", "N/A"]

export interface ReleasesListOpts {
	readonly serviceName?: string
	readonly serviceNames?: readonly string[]
	readonly environments?: readonly string[]
	readonly namespaces?: readonly string[]
	readonly excludedEnvironments?: readonly string[]
	readonly excludedNamespaces?: readonly string[]
	readonly limit?: number
}

export interface ReleasesListOutput {
	readonly serviceName: string
	readonly environment: string
	readonly commitSha: string
	readonly firstSeen: DateTime.Utc
	readonly spanCount: number
	readonly errorCount: number
	readonly p50LatencyMs: number
	readonly p95LatencyMs: number
	readonly p99LatencyMs: number
	readonly apdexSatisfiedCount: number
	readonly apdexToleratingCount: number
}

export const releasesListRowSchema = Schema.Struct({
	serviceName: Schema.String,
	environment: Schema.String,
	commitSha: Schema.String,
	firstSeen: DateTimeUtcFromWarehouse,
	// `CHNumber`, never `Schema.Number`: UInt64 counts arrive quoted on a
	// gateway that refuses `output_format_json_quote_64bit_integers=0`.
	spanCount: CHNumber,
	errorCount: CHNumber,
	p50LatencyMs: CHNumber,
	p95LatencyMs: CHNumber,
	p99LatencyMs: CHNumber,
	apdexSatisfiedCount: CHNumber,
	apdexToleratingCount: CHNumber,
}) satisfies CompiledQueryRowSchema<ReleasesListOutput>

/**
 * One row per (service, environment, commit) in the window, newest first.
 *
 * Every version of a service is a row, not just the latest — a release's
 * impact is "this version against every other version of the same service in
 * the same minutes", and the caller derives that split from these rows. The
 * same query scoped to one service is therefore also the detail page's
 * comparison table.
 */
export function releasesListQuery(opts: ReleasesListOpts = {}) {
	return serviceOverviewWindows({
		serviceName: opts.serviceName,
		environments: opts.environments,
		namespaces: opts.namespaces,
		excludedEnvironments: opts.excludedEnvironments,
		excludedNamespaces: opts.excludedNamespaces,
	})
		.select(($) => ({
			serviceName: $.bServiceName,
			environment: $.bEnvironment,
			commitSha: $.bCommitSha,
			firstSeen: CH.min_($.bFirstSeen),
			spanCount: CH.sum($.bSpanCount),
			errorCount: CH.sum($.bErrorCount),
			p50LatencyMs: CH.rawExpr(
				"arrayElement(quantilesTDigestMerge(0.5, 0.95, 0.99)(bDurationQuantiles), 1) / 1000000",
				T.float64,
			),
			p95LatencyMs: CH.rawExpr(
				"arrayElement(quantilesTDigestMerge(0.5, 0.95, 0.99)(bDurationQuantiles), 2) / 1000000",
				T.float64,
			),
			p99LatencyMs: CH.rawExpr(
				"arrayElement(quantilesTDigestMerge(0.5, 0.95, 0.99)(bDurationQuantiles), 3) / 1000000",
				T.float64,
			),
			apdexSatisfiedCount: CH.sum($.bApdexSatisfiedCount),
			apdexToleratingCount: CH.sum($.bApdexToleratingCount),
		}))
		.where(($) => [
			CH.notInList($.bCommitSha, PLACEHOLDER_COMMIT_SHAS),
			opts.serviceNames?.length ? CH.inList($.bServiceName, opts.serviceNames) : undefined,
		])
		.groupBy("serviceName", "environment", "commitSha")
		.orderBy(["firstSeen", "desc"], ["spanCount", "desc"])
		.limit(opts.limit ?? RELEASES_LIST_CAP)
		.format("JSON")
}

// Running commits
//
// Which commit each service served traffic from in the window, for the dependency scanner: a
// vulnerable package matters most in the commit a busy service is running.

/** At most this many (service, environment, commit) rows, busiest first. */
export const RUNNING_COMMITS_CAP = 200

export interface RunningCommitsOutput {
	readonly serviceName: string
	readonly environment: string
	readonly commitSha: string
	readonly spanCount: number
}

export const runningCommitsRowSchema = Schema.Struct({
	serviceName: Schema.String,
	environment: Schema.String,
	commitSha: Schema.String,
	spanCount: CHNumber,
}) satisfies CompiledQueryRowSchema<RunningCommitsOutput>

/** The releases list without its latency and error columns, ordered by traffic. */
export function runningCommitsQuery() {
	return serviceOverviewWindows({})
		.select(($) => ({
			serviceName: $.bServiceName,
			environment: $.bEnvironment,
			commitSha: $.bCommitSha,
			spanCount: CH.sum($.bSpanCount),
		}))
		.where(($) => [CH.notInList($.bCommitSha, PLACEHOLDER_COMMIT_SHAS)])
		.groupBy("serviceName", "environment", "commitSha")
		.orderBy(["spanCount", "desc"])
		.limit(RUNNING_COMMITS_CAP)
		.format("JSON")
}

// Releases timeline
//
// The org-wide sibling of `serviceReleasesTimelineQuery`: per bucket, per
// service, per commit. Feeds the swimlanes and the per-service rollout share
// (which version carried the last bucket's traffic).

export interface ReleasesTimelineOpts {
	readonly serviceName?: string
	readonly serviceNames?: readonly string[]
	readonly environments?: readonly string[]
	readonly namespaces?: readonly string[]
	readonly excludedEnvironments?: readonly string[]
	readonly excludedNamespaces?: readonly string[]
	/**
	 * Needed at build time, not just as a compile parameter: it selects which
	 * rollup tiers can answer, because a tier coarser than the bucket has no
	 * position inside it.
	 */
	readonly bucketSeconds: number
}

export interface ReleasesTimelineOutput {
	readonly bucket: DateTime.Utc
	readonly serviceName: string
	readonly commitSha: string
	readonly count: number
}

const RELEASES_TIMELINE_CAP = 5000

/** Sub-minute buckets: no rollup tier can place a row inside a minute. */
function releasesTimelineRawQuery(
	opts: ReleasesTimelineOpts,
): CHQuery<ColumnDefs, ReleasesTimelineOutput, {}> {
	return from(ServiceOverviewSpans)
		.select(($) => ({
			bucket: CH.toStartOfInterval($.Timestamp, param.int("bucketSeconds")),
			serviceName: $.ServiceName,
			commitSha: $.CommitSha,
			count: CH.count(),
		}))
		.where(($) => [
			...serviceOverviewWhereConditions($, {
				serviceName: opts.serviceName,
				serviceNames: opts.serviceNames,
				environments: opts.environments,
				namespaces: opts.namespaces,
				excludedEnvironments: opts.excludedEnvironments,
				excludedNamespaces: opts.excludedNamespaces,
			}),
			CH.notInList($.CommitSha, PLACEHOLDER_COMMIT_SHAS),
		])
		.groupBy("bucket", "serviceName", "commitSha")
		.orderBy(["bucket", "asc"])
		.limit(RELEASES_TIMELINE_CAP)
		.format("JSON") as CHQuery<ColumnDefs, ReleasesTimelineOutput, {}>
}

export function releasesTimelineQuery(
	opts: ReleasesTimelineOpts,
): CHQuery<ColumnDefs, ReleasesTimelineOutput, {}> {
	const tiers = serviceWindowTiersForBucket(opts.bucketSeconds)
	if (tiers === "raw") return releasesTimelineRawQuery(opts)

	return serviceOverviewWindows(
		{
			serviceName: opts.serviceName,
			environments: opts.environments,
			namespaces: opts.namespaces,
			excludedEnvironments: opts.excludedEnvironments,
			excludedNamespaces: opts.excludedNamespaces,
		},
		tiers,
	)
		.select(($) => ({
			bucket: CH.toStartOfInterval($.bBucket, param.int("bucketSeconds")),
			serviceName: $.bServiceName,
			commitSha: $.bCommitSha,
			count: CH.sum($.bSpanCount),
		}))
		.where(($) => [
			CH.notInList($.bCommitSha, PLACEHOLDER_COMMIT_SHAS),
			opts.serviceNames?.length ? CH.inList($.bServiceName, opts.serviceNames) : undefined,
		])
		.groupBy("bucket", "serviceName", "commitSha")
		.orderBy(["bucket", "asc"])
		.limit(RELEASES_TIMELINE_CAP)
		.format("JSON") as CHQuery<ColumnDefs, ReleasesTimelineOutput, {}>
}

// Error fingerprints on a version
//
// The bridge from a release to the issues system. Traces key a release on
// `vcs.ref.head.revision`; error events key on `service.version`, which Maple's
// SDKs stamp with the same sha. Reads the per-occurrence projection because a
// release starts at an arbitrary instant and a minute rollup would smear the
// first occurrences across the deploy boundary.

export interface ReleaseErrorFingerprintsOpts {
	readonly serviceName: string
	readonly environments?: readonly string[]
	readonly limit?: number
}

export interface ReleaseErrorFingerprintsOutput {
	readonly fingerprintHash: string
	readonly count: number
	readonly firstSeen: DateTime.Utc
}

export function releaseErrorFingerprintsQuery(opts: ReleaseErrorFingerprintsOpts) {
	return from(ErrorEventsByTime)
		.select(($) => ({
			fingerprintHash: CH.toString_($.FingerprintHash),
			count: CH.count(),
			firstSeen: CH.min_($.Timestamp),
		}))
		.where(($) => [
			$.OrgId.eq(orgIdParam),
			$.ServiceName.eq(opts.serviceName),
			$.ServiceVersion.eq(param.string("serviceVersion")),
			$.Timestamp.gte(utcSecondsParam("startTime")),
			$.Timestamp.lte(utcSecondsParam("endTime")),
			opts.environments?.length ? CH.inList($.DeploymentEnv, opts.environments) : undefined,
		])
		.groupBy("fingerprintHash")
		.orderBy(["count", "desc"])
		.limit(opts.limit ?? 50)
		.format("JSON")
}

// Service deployments
//
// The releases list plus when each version last served, for deploy
// verification in one call. `lastSeen` is the start of the last bucket the
// version had traffic in: minute precision without the hourly tier, hour
// precision for hours wholly inside the window with it.

export interface ServiceDeploymentsOpts {
	readonly serviceName?: string
	readonly environments?: readonly string[]
	/** Read minutely buckets for the whole window so `lastSeen` is minute-exact. */
	readonly minutePrecision: boolean
	readonly limit?: number
	/** Versions kept per (service, environment), most recently serving first. */
	readonly perServiceLimit?: number
}

export interface ServiceDeploymentsOutput extends ReleasesListOutput {
	readonly lastSeen: DateTime.Utc
}

export const serviceDeploymentsRowSchema = Schema.Struct({
	...releasesListRowSchema.fields,
	lastSeen: DateTimeUtcFromWarehouse,
}) satisfies CompiledQueryRowSchema<ServiceDeploymentsOutput>

export const DEPLOYMENTS_PER_SERVICE_CAP = 20

type DeploymentKey = keyof ServiceDeploymentsOutput

const deploymentColumns = <A extends { readonly [K in DeploymentKey]: unknown }>(
	$: A,
): Pick<A, DeploymentKey> => ({
	serviceName: $.serviceName,
	environment: $.environment,
	commitSha: $.commitSha,
	firstSeen: $.firstSeen,
	lastSeen: $.lastSeen,
	spanCount: $.spanCount,
	errorCount: $.errorCount,
	p50LatencyMs: $.p50LatencyMs,
	p95LatencyMs: $.p95LatencyMs,
	p99LatencyMs: $.p99LatencyMs,
	apdexSatisfiedCount: $.apdexSatisfiedCount,
	apdexToleratingCount: $.apdexToleratingCount,
})

export function serviceDeploymentsQuery(opts: ServiceDeploymentsOpts) {
	const versions = serviceOverviewWindows(
		{ serviceName: opts.serviceName, environments: opts.environments },
		{ grain: "minute", includeHourly: !opts.minutePrecision },
	)
		.select(($) => ({
			serviceName: $.bServiceName,
			environment: $.bEnvironment,
			commitSha: $.bCommitSha,
			firstSeen: CH.min_($.bFirstSeen),
			lastSeen: CH.max_($.bBucket),
			spanCount: CH.sum($.bSpanCount),
			errorCount: CH.sum($.bErrorCount),
			p50LatencyMs: CH.rawExpr(
				"arrayElement(quantilesTDigestMerge(0.5, 0.95, 0.99)(bDurationQuantiles), 1) / 1000000",
				T.float64,
			),
			p95LatencyMs: CH.rawExpr(
				"arrayElement(quantilesTDigestMerge(0.5, 0.95, 0.99)(bDurationQuantiles), 2) / 1000000",
				T.float64,
			),
			p99LatencyMs: CH.rawExpr(
				"arrayElement(quantilesTDigestMerge(0.5, 0.95, 0.99)(bDurationQuantiles), 3) / 1000000",
				T.float64,
			),
			apdexSatisfiedCount: CH.sum($.bApdexSatisfiedCount),
			apdexToleratingCount: CH.sum($.bApdexToleratingCount),
		}))
		.where(($) => [CH.notInList($.bCommitSha, PLACEHOLDER_COMMIT_SHAS)])
		.groupBy("serviceName", "environment", "commitSha")

	// Rank inside each (service, environment) so one fast-deploying service
	// cannot crowd the others out of the global limit.
	const ranked = fromQuery(versions, "versions").select(($) => ({
		...deploymentColumns($),
		versionRank: CH.rawExpr(
			"row_number() OVER (PARTITION BY serviceName, environment ORDER BY lastSeen DESC, firstSeen DESC)",
			T.uint64,
		),
	}))

	return fromQuery(ranked, "ranked")
		.select(($) => deploymentColumns($))
		.where(($) => [$.versionRank.lte(opts.perServiceLimit ?? DEPLOYMENTS_PER_SERVICE_CAP)])
		.orderBy(["serviceName", "asc"], ["firstSeen", "desc"])
		.limit(opts.limit ?? RELEASES_LIST_CAP)
		.format("JSON")
}