builders.rs

use std::collections::BTreeMap;

use k8s_openapi::{
	api::{
		apps::v1::{Deployment, DeploymentSpec},
		batch::v1::{Job, JobSpec},
		core::v1::{
			Container, ContainerPort, EnvVar, ExecAction, PersistentVolumeClaim,
			PersistentVolumeClaimSpec, PodSpec, PodTemplateSpec, Probe, ResourceRequirements,
			SecretReference, Volume, VolumeMount, VolumeResourceRequirements,
		},
	},
	apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::LabelSelector},
};
use kube::{ResourceExt, api::ObjectMeta};
use kube_quantity::ParsedQuantity;
use rust_decimal::Decimal;

use super::restore_owner_reference;
use crate::{
	controllers::{env_from_secret, env_from_secret_optional, kopia_writable_env},
	error::{Error, Result},
	quantity::compute_shm_and_shared_buffers,
	types::*,
};

/// Name of the credential-reset Job for a given restore.
pub fn credential_reset_job_name(restore_name: &str) -> String {
	format!("{restore_name}-cred-reset")
}

/// Build a Job that resets the analytics user's password on a restore whose
/// Postgres deployment has been scaled to zero.
///
/// The job mounts the restore's data PVC directly, starts a temporary
/// `postgres --single` process (no TCP listener, no auth), runs the
/// ALTER ROLE statement, then exits. The deployment must already be scaled
/// to 0 before this job is created so that the PVC is not in use.
pub fn build_credential_reset_job(
	restore: &PostgresPhysicalRestore,
	replica: &PostgresPhysicalReplica,
	job_name: &str,
	namespace: &str,
) -> Result<Job> {
	let pvc_name = format!("{}-data", restore.name_any());

	let pg_version = restore
		.status
		.as_ref()
		.and_then(|s| s.postgres_version.as_ref())
		.cloned()
		.ok_or_else(|| Error::MissingField("status.postgresVersion".to_string()))?;

	let pg_image = format!("postgres:{pg_version}");

	let creds_secret = SecretReference {
		name: Some(format!("{}-creds", restore.spec.replica.name)),
		namespace: Some(namespace.to_string()),
	};

	// ANALYTICS_PASSWORD is operator-generated (not user input), so direct
	// shell interpolation into the SQL string is safe.
	let script = r#"set -e
PGDATA=/pgdata/pgdata

echo "Resetting analytics user password via single-user mode..."
echo "ALTER ROLE ${ANALYTICS_USERNAME} WITH PASSWORD '${ANALYTICS_PASSWORD}';" \
  | postgres --single -D "$PGDATA" postgres

echo "Credential reset complete."
"#
	.to_string();

	let labels = BTreeMap::from([
		(
			"pgro.bes.au/replica".to_string(),
			restore.spec.replica.name.clone(),
		),
		("pgro.bes.au/restore".to_string(), restore.name_any()),
		("pgro.bes.au/job-type".to_string(), "cred-reset".to_string()),
	]);

	Ok(Job {
		metadata: ObjectMeta {
			name: Some(job_name.to_string()),
			namespace: Some(namespace.to_string()),
			labels: Some(labels.clone()),
			owner_references: Some(vec![restore_owner_reference(restore)]),
			..Default::default()
		},
		spec: Some(JobSpec {
			backoff_limit: Some(2),
			active_deadline_seconds: Some(120),
			ttl_seconds_after_finished: Some(600),
			template: PodTemplateSpec {
				metadata: Some(ObjectMeta {
					labels: Some(labels),
					..Default::default()
				}),
				spec: Some(PodSpec {
					restart_policy: Some("Never".to_string()),
					security_context: Some(k8s_openapi::api::core::v1::PodSecurityContext {
						run_as_user: Some(999),
						run_as_group: Some(999),
						fs_group: Some(999),
						..Default::default()
					}),
					containers: vec![Container {
						name: "cred-reset".to_string(),
						image: Some(pg_image),
						command: Some(vec!["/bin/sh".to_string(), "-c".to_string()]),
						args: Some(vec![script]),
						env: Some(vec![
							EnvVar {
								name: "ANALYTICS_USERNAME".to_string(),
								value: Some(replica.spec.analytics_username.clone()),
								..Default::default()
							},
							env_from_secret("ANALYTICS_PASSWORD", &creds_secret, "password"),
						]),
						volume_mounts: Some(vec![VolumeMount {
							name: "pgdata".to_string(),
							mount_path: "/pgdata".to_string(),
							..Default::default()
						}]),
						resources: Some(ResourceRequirements {
							requests: Some(BTreeMap::from([
								("cpu".to_string(), Quantity("100m".to_string())),
								("memory".to_string(), Quantity("128Mi".to_string())),
							])),
							limits: Some(BTreeMap::from([
								("cpu".to_string(), Quantity("500m".to_string())),
								("memory".to_string(), Quantity("256Mi".to_string())),
							])),
							..Default::default()
						}),
						..Default::default()
					}],
					volumes: Some(vec![Volume {
						name: "pgdata".to_string(),
						persistent_volume_claim: Some(
							k8s_openapi::api::core::v1::PersistentVolumeClaimVolumeSource {
								claim_name: pvc_name,
								read_only: Some(false),
							},
						),
						..Default::default()
					}]),
					affinity: replica.spec.affinity.clone(),
					tolerations: Some(replica.spec.tolerations.clone()),
					..Default::default()
				}),
			},
			..Default::default()
		}),
		..Default::default()
	})
}

pub fn build_version_detect_job(
	restore: &PostgresPhysicalRestore,
	job_name: &str,
	namespace: &str,
	pvc_name: &str,
) -> Job {
	let script = r#"set -e

echo "PVC contents:"
ls -la /pgdata/ 2>&1 || true
echo "---"
ls -la /pgdata/pgdata/ 2>&1 || true
echo "---"

# If the pgdata symlink already exists, just read the version
if [ -L /pgdata/pgdata ] && [ -f /pgdata/pgdata/PG_VERSION ]; then
  VERSION=$(cat /pgdata/pgdata/PG_VERSION)
  echo "Detected postgres version: $VERSION"
  echo -n "$VERSION" > /dev/termination-log
  exit 0
fi

# Otherwise locate PGDATA and recreate the symlink
echo "pgdata symlink missing, locating PGDATA directory..."
PGDATA_DIR=""

# Prefer 'current' symlink (org convention)
if [ -L /pgdata/postgres/current ]; then
  LINK_TARGET=$(readlink /pgdata/postgres/current)
  RELATIVE=$(echo "$LINK_TARGET" | sed 's|.*/\([0-9]\{1,\}/\)|/pgdata/postgres/\1|')
  if [ -f "$RELATIVE/PG_VERSION" ]; then
    PGDATA_DIR="$RELATIVE"
    echo "Found PGDATA via 'current' symlink: $PGDATA_DIR"
  fi
fi

# Fallback: pick the highest version directory containing PG_VERSION
# Filter to cluster-root directories only (must contain a 'global' subdirectory)
if [ -z "$PGDATA_DIR" ]; then
  PGDATA_DIR=$(find /pgdata/postgres -name "PG_VERSION" 2>/dev/null | while read -r f; do
    dir=$(dirname "$f")
    [ -d "$dir/global" ] && echo "$dir"
  done | sort -t/ -k4 -rn | head -1)
fi

# Last resort: search anywhere under /pgdata
if [ -z "$PGDATA_DIR" ]; then
  echo "Searching for PG_VERSION recursively..."
  find /pgdata -name "PG_VERSION" 2>/dev/null || true
  PGDATA_DIR=$(find /pgdata -name "PG_VERSION" 2>/dev/null | while read -r f; do
    dir=$(dirname "$f")
    [ -d "$dir/global" ] && echo "$dir"
  done | sort -t/ -k4 -rn | head -1)
fi

if [ -z "$PGDATA_DIR" ]; then
  echo "ERROR: Could not detect postgres version from PVC"
  exit 1
fi

echo "Found PGDATA at: $PGDATA_DIR"
ln -sfn "$PGDATA_DIR" /pgdata/pgdata

VERSION=$(cat /pgdata/pgdata/PG_VERSION)
echo "Detected postgres version: $VERSION"
echo "$VERSION" > /pgdata/.postgres-version
echo -n "$VERSION" > /dev/termination-log
"#;

	Job {
		metadata: ObjectMeta {
			name: Some(job_name.to_string()),
			namespace: Some(namespace.to_string()),
			labels: Some(BTreeMap::from([
				(
					"pgro.bes.au/replica".to_string(),
					restore.spec.replica.name.clone(),
				),
				("pgro.bes.au/restore".to_string(), restore.name_any()),
				(
					"pgro.bes.au/job-type".to_string(),
					"version-detect".to_string(),
				),
			])),
			owner_references: Some(vec![restore_owner_reference(restore)]),
			..Default::default()
		},
		spec: Some(JobSpec {
			backoff_limit: Some(2),
			active_deadline_seconds: Some(120),
			ttl_seconds_after_finished: Some(120),
			template: PodTemplateSpec {
				metadata: Some(ObjectMeta {
					labels: Some(BTreeMap::from([
						(
							"pgro.bes.au/replica".to_string(),
							restore.spec.replica.name.clone(),
						),
						("pgro.bes.au/restore".to_string(), restore.name_any()),
					])),
					..Default::default()
				}),
				spec: Some(PodSpec {
					restart_policy: Some("Never".to_string()),
					security_context: Some(k8s_openapi::api::core::v1::PodSecurityContext {
						run_as_user: Some(999),
						run_as_group: Some(999),
						fs_group: Some(999),
						..Default::default()
					}),
					containers: vec![Container {
						name: "version-detect".to_string(),
						image: Some("alpine:latest".to_string()),
						command: Some(vec!["/bin/sh".to_string(), "-c".to_string()]),
						args: Some(vec![script.to_string()]),
						volume_mounts: Some(vec![VolumeMount {
							name: "pgdata".to_string(),
							mount_path: "/pgdata".to_string(),
							..Default::default()
						}]),
						resources: Some(ResourceRequirements {
							requests: Some(BTreeMap::from([
								("cpu".to_string(), Quantity("10m".to_string())),
								("memory".to_string(), Quantity("16Mi".to_string())),
							])),
							limits: Some(BTreeMap::from([
								("cpu".to_string(), Quantity("50m".to_string())),
								("memory".to_string(), Quantity("32Mi".to_string())),
							])),
							..Default::default()
						}),
						..Default::default()
					}],
					volumes: Some(vec![Volume {
						name: "pgdata".to_string(),
						persistent_volume_claim: Some(
							k8s_openapi::api::core::v1::PersistentVolumeClaimVolumeSource {
								claim_name: pvc_name.to_string(),
								read_only: Some(false),
							},
						),
						..Default::default()
					}]),
					..Default::default()
				}),
			},
			..Default::default()
		}),
		..Default::default()
	}
}

pub fn build_pvc(
	restore: &PostgresPhysicalRestore,
	pvc_name: &str,
	namespace: &str,
	replica: &PostgresPhysicalReplica,
) -> Result<PersistentVolumeClaim> {
	Ok(PersistentVolumeClaim {
		metadata: ObjectMeta {
			name: Some(pvc_name.to_string()),
			namespace: Some(namespace.to_string()),
			labels: Some(BTreeMap::from([
				(
					"pgro.bes.au/replica".to_string(),
					restore.spec.replica.name.clone(),
				),
				("pgro.bes.au/restore".to_string(), restore.name_any()),
			])),
			owner_references: Some(vec![restore_owner_reference(restore)]),
			..Default::default()
		},
		spec: Some(PersistentVolumeClaimSpec {
			access_modes: Some(vec!["ReadWriteOnce".to_string()]),
			storage_class_name: replica.spec.storage_class.clone(),
			resources: Some(VolumeResourceRequirements {
				requests: Some(BTreeMap::from([(
					"storage".to_string(),
					restore.spec.storage_size.clone(),
				)])),
				..Default::default()
			}),
			..Default::default()
		}),
		..Default::default()
	})
}

/// Name of the per-replica kopia cache PVC. One per replica, reused across
/// every restore Job for that replica — kopia's content cache then survives
/// restore-to-restore and the next snapshot only has to download new
/// blobs. Owned by the replica (cascade-deleted with it), unlike the
/// per-restore data PVC.
pub fn kopia_cache_pvc_name(replica_name: &str) -> String {
	format!("{replica_name}-kopia-cache")
}

/// Returns true if `desired` is strictly larger than `current`. Used by
/// the restore controller to apply ratchet semantics to the cache PVC:
/// grow on snapshot growth, never shrink. Returns false on parse error
/// so a corrupt or unparsable quantity doesn't cause spurious resize
/// attempts.
pub fn cache_size_needs_grow(current: &Quantity, desired: &Quantity) -> bool {
	match (
		ParsedQuantity::try_from(current.clone()),
		ParsedQuantity::try_from(desired.clone()),
	) {
		(Ok(c), Ok(d)) => d > c,
		_ => false,
	}
}

/// Compute the per-replica kopia cache PVC size as `max(10Gi, 20% of
/// snapshot size)`. Kopia caches snapshot metadata, indices, and content
/// blobs; sizing relative to the snapshot scales naturally with the data
/// volume, and the 10Gi floor catches tiny snapshots where 20% would
/// leave no room for incremental churn.
pub fn kopia_cache_pvc_size(snapshot_size: &Quantity) -> Quantity {
	let twenty_percent = ParsedQuantity::try_from(snapshot_size.clone())
		.map(|q| q * Decimal::new(2, 1))
		.unwrap_or_else(|_| ParsedQuantity::from(Decimal::ZERO));
	let floor = ParsedQuantity::try_from("10Gi").expect("10Gi parses");
	let chosen = if twenty_percent > floor {
		twenty_percent
	} else {
		floor
	};
	chosen.into()
}

pub fn build_kopia_cache_pvc(
	replica: &PostgresPhysicalReplica,
	snapshot_size: &Quantity,
	namespace: &str,
) -> PersistentVolumeClaim {
	let replica_name = replica.name_any();
	PersistentVolumeClaim {
		metadata: ObjectMeta {
			name: Some(kopia_cache_pvc_name(&replica_name)),
			namespace: Some(namespace.to_string()),
			labels: Some(BTreeMap::from([
				("pgro.bes.au/replica".to_string(), replica_name),
				(
					"pgro.bes.au/component".to_string(),
					"kopia-cache".to_string(),
				),
			])),
			owner_references: Some(vec![replica.owner_reference()]),
			..Default::default()
		},
		spec: Some(PersistentVolumeClaimSpec {
			access_modes: Some(vec!["ReadWriteOnce".to_string()]),
			storage_class_name: replica.spec.storage_class.clone(),
			resources: Some(VolumeResourceRequirements {
				requests: Some(BTreeMap::from([(
					"storage".to_string(),
					kopia_cache_pvc_size(snapshot_size),
				)])),
				..Default::default()
			}),
			..Default::default()
		}),
		..Default::default()
	}
}

pub fn build_restore_job(
	restore: &PostgresPhysicalRestore,
	job_name: &str,
	namespace: &str,
	replica: &PostgresPhysicalReplica,
	kopia_image: &str,
) -> Result<Job> {
	let kopia_secret = &replica.spec.kopia_secret_ref;
	let pvc_name = format!("{}-data", restore.name_any());
	let cache_pvc_name = kopia_cache_pvc_name(&restore.spec.replica.name);

	let restore_script = r#"set -e

mkdir -p /tmp/kopia/config /tmp/kopia/logs /tmp/kopia/cache

ENDPOINT_ARGS=""
if [ -n "$KOPIA_ENDPOINT" ]; then
  ENDPOINT_ARGS="--endpoint=$KOPIA_ENDPOINT"
fi
if [ "$KOPIA_DISABLE_TLS" = "true" ]; then
  ENDPOINT_ARGS="$ENDPOINT_ARGS --disable-tls --disable-tls-verification"
fi

echo "Connecting to kopia repository..."
kopia repository connect s3 \
  --bucket="$KOPIA_BUCKET" \
  --region="$KOPIA_REGION" \
  --access-key="$AWS_ACCESS_KEY_ID" \
  --secret-access-key="$AWS_SECRET_ACCESS_KEY" \
  --password="$KOPIA_PASSWORD" \
  $ENDPOINT_ARGS

echo "Starting restore..."
kopia snapshot restore "$SNAPSHOT_ID" /pgdata/postgres

echo "Restore complete"
ls -la /pgdata/

echo "Locating PGDATA directory..."

# Prefer the 'current' symlink if it exists (org convention)
if [ -L /pgdata/postgres/current ]; then
  # The symlink target is an absolute path from the original host, resolve it
  # relative to /pgdata/postgres by extracting the version/cluster part.
  LINK_TARGET=$(readlink /pgdata/postgres/current)
  # e.g. /var/lib/postgresql/16/main -> try /pgdata/postgres/16/main
  RELATIVE=$(echo "$LINK_TARGET" | sed 's|.*/\([0-9]\{1,\}/\)|/pgdata/postgres/\1|')
  if [ -f "$RELATIVE/PG_VERSION" ]; then
    PGDATA_DIR="$RELATIVE"
    echo "Found PGDATA via 'current' symlink: $PGDATA_DIR"
  fi
fi

# Fallback: pick the highest version directory containing PG_VERSION
# Filter to cluster-root directories only (must contain a 'global' subdirectory)
if [ -z "$PGDATA_DIR" ]; then
  PGDATA_DIR=$(find /pgdata/postgres -name "PG_VERSION" 2>/dev/null | while read -r f; do
    dir=$(dirname "$f")
    [ -d "$dir/global" ] && echo "$dir"
  done | sort -t/ -k4 -rn | head -1)
fi

if [ -z "$PGDATA_DIR" ]; then
  echo "ERROR: Could not find PG_VERSION in restored data"
  exit 1
fi
echo "Found PGDATA at: $PGDATA_DIR"
ln -sfn "$PGDATA_DIR" /pgdata/pgdata
rm -f "$PGDATA_DIR/postmaster.pid"

VERSION=$(cat /pgdata/pgdata/PG_VERSION)
echo "Detected postgres version: $VERSION"
echo "$VERSION" > /pgdata/.postgres-version
echo -n "$VERSION" > /dev/termination-log
"#;

	Ok(Job {
		metadata: ObjectMeta {
			name: Some(job_name.to_string()),
			namespace: Some(namespace.to_string()),
			labels: Some(BTreeMap::from([
				(
					"pgro.bes.au/replica".to_string(),
					restore.spec.replica.name.clone(),
				),
				("pgro.bes.au/restore".to_string(), restore.name_any()),
			])),
			owner_references: Some(vec![restore_owner_reference(restore)]),
			..Default::default()
		},
		spec: Some(JobSpec {
			backoff_limit: Some(3),
			active_deadline_seconds: Some(7200),   // 2 hours
			ttl_seconds_after_finished: Some(120), // safety net if operator misses deletion
			template: PodTemplateSpec {
				metadata: Some(ObjectMeta {
					labels: Some(BTreeMap::from([
						(
							"pgro.bes.au/replica".to_string(),
							restore.spec.replica.name.clone(),
						),
						("pgro.bes.au/restore".to_string(), restore.name_any()),
					])),
					..Default::default()
				}),
				spec: Some(PodSpec {
					restart_policy: Some("Never".to_string()),
					security_context: Some(k8s_openapi::api::core::v1::PodSecurityContext {
						run_as_user: Some(999),
						run_as_group: Some(999),
						fs_group: Some(999),
						..Default::default()
					}),

					containers: vec![Container {
						name: "restore".to_string(),
						image: Some(kopia_image.to_string()),
						command: Some(vec!["/bin/sh".to_string(), "-c".to_string()]),
						args: Some(vec![restore_script.to_string()]),
						env: Some(
							[
								vec![EnvVar {
									name: "SNAPSHOT_ID".to_string(),
									value: Some(restore.spec.snapshot.clone()),
									..Default::default()
								}],
								kopia_writable_env(),
								vec![
									env_from_secret("KOPIA_BUCKET", kopia_secret, "bucket"),
									env_from_secret("KOPIA_REGION", kopia_secret, "region"),
									env_from_secret(
										"AWS_ACCESS_KEY_ID",
										kopia_secret,
										"accessKeyId",
									),
									env_from_secret(
										"AWS_SECRET_ACCESS_KEY",
										kopia_secret,
										"secretAccessKey",
									),
									env_from_secret(
										"KOPIA_PASSWORD",
										kopia_secret,
										"repositoryPassword",
									),
									env_from_secret_optional(
										"KOPIA_ENDPOINT",
										kopia_secret,
										"endpoint",
									),
									env_from_secret_optional(
										"KOPIA_DISABLE_TLS",
										kopia_secret,
										"disableTls",
									),
								],
							]
							.concat(),
						),
						volume_mounts: Some(vec![
							VolumeMount {
								name: "pgdata".to_string(),
								mount_path: "/pgdata".to_string(),
								..Default::default()
							},
							k8s_openapi::api::core::v1::VolumeMount {
								name: "kopia-cache".to_string(),
								mount_path: "/tmp/kopia".to_string(),
								..Default::default()
							},
						]),
						resources: Some(ResourceRequirements {
							requests: Some(BTreeMap::from([
								("cpu".to_string(), Quantity("500m".to_string())),
								("memory".to_string(), Quantity("1Gi".to_string())),
							])),
							limits: Some(BTreeMap::from([
								("cpu".to_string(), Quantity("2".to_string())),
								("memory".to_string(), Quantity("4Gi".to_string())),
							])),
							..Default::default()
						}),
						..Default::default()
					}],
					volumes: Some(vec![
						Volume {
							name: "pgdata".to_string(),
							persistent_volume_claim: Some(
								k8s_openapi::api::core::v1::PersistentVolumeClaimVolumeSource {
									claim_name: pvc_name,
									read_only: Some(false),
								},
							),
							..Default::default()
						},
						Volume {
							name: "kopia-cache".to_string(),
							persistent_volume_claim: Some(
								k8s_openapi::api::core::v1::PersistentVolumeClaimVolumeSource {
									claim_name: cache_pvc_name,
									read_only: Some(false),
								},
							),
							..Default::default()
						},
					]),
					..Default::default()
				}),
			},
			..Default::default()
		}),
		..Default::default()
	})
}

pub fn build_deployment(
	restore: &PostgresPhysicalRestore,
	name: &str,
	namespace: &str,
	replica: &PostgresPhysicalReplica,
) -> Result<Deployment> {
	let pvc_name = format!("{name}-data");
	let (shm_size, shared_buffers_mb) = compute_shm_and_shared_buffers(&replica.spec.resources);
	let creds_secret = SecretReference {
		name: Some(format!("{}-creds", restore.spec.replica.name)),
		namespace: Some(namespace.to_string()),
	};

	let pg_version = restore
		.status
		.as_ref()
		.and_then(|s| s.postgres_version.as_ref())
		.cloned()
		.ok_or_else(|| Error::MissingField("status.postgresVersion".to_string()))?;

	let pg_image = format!("postgres:{pg_version}");

	let locale_script = r#"set -ex
echo "Creating Windows-compatible locales..."
for lang in \
  "English_United States" \
  "English_United Kingdom" \
; do
  localedef -i en_US -f CP1250 "${lang}.1250" 2>/dev/null || true
  localedef -i en_US -f CP1251 "${lang}.1251" 2>/dev/null || true
  localedef -i en_US -f CP1252 "${lang}.1252" 2>/dev/null || true
  localedef -i en_US -f CP1253 "${lang}.1253" 2>/dev/null || true
  localedef -i en_US -f CP1254 "${lang}.1254" 2>/dev/null || true
  localedef -i en_US -f CP1255 "${lang}.1255" 2>/dev/null || true
  localedef -i en_US -f CP1256 "${lang}.1256" 2>/dev/null || true
  localedef -i en_US -f CP1257 "${lang}.1257" 2>/dev/null || true
  localedef -i en_US -f CP1258 "${lang}.1258" 2>/dev/null || true
  localedef -i en_US -f UTF-8  "${lang}.65001" 2>/dev/null || true
done
echo "Copying locale data to shared volume..."
cp -a /usr/lib/locale/* /locale-data/
"#
	.to_string();

	// persistent_schemas needs write access to receive the migrated data
	let effective_read_only = replica.spec.read_only && replica.spec.persistent_schemas.is_none();
	let read_only = effective_read_only.to_string();

	let extra_config_block = if let Some(ref extra) = replica.spec.postgres_extra_config {
		format!(
			r#"echo "Appending extra postgresql.conf settings..."
cat >> "$PGDATA/postgresql.conf" << 'EXTRACONFEOF'
{extra}
EXTRACONFEOF"#
		)
	} else {
		String::new()
	};

	let init_script = format!(
		r#"set -ex
PGDATA=/pgdata/pgdata

chmod 0750 "$PGDATA"

if [ ! -f "$PGDATA/postgresql.conf" ]; then
  echo "Creating minimal postgresql.conf (Debian-style installs keep config in /etc)..."
  cat > "$PGDATA/postgresql.conf" << 'CONFEOF'
listen_addresses = '*'
port = 5432
max_connections = 100
max_prepared_transactions = 16
shared_buffers = {shared_buffers_mb}MB
dynamic_shared_memory_type = posix
log_timezone = 'UTC'
datestyle = 'iso, mdy'
timezone = 'UTC'
lc_messages = 'C'
lc_monetary = 'C'
lc_numeric = 'C'
lc_time = 'C'
CONFEOF
fi

{extra_config_block}

echo "Stripping source-host config overrides from postgresql.conf..."
sed -i \
  -e '/^[[:space:]]*hba_file[[:space:]]*=/d' \
  -e '/^[[:space:]]*ident_file[[:space:]]*=/d' \
  -e '/^[[:space:]]*data_directory[[:space:]]*=/d' \
  -e '/^[[:space:]]*dynamic_shared_memory_type[[:space:]]*=/d' \
  -e '/^[[:space:]]*shared_buffers[[:space:]]*=/d' \
  -e '/^[[:space:]]*log_destination[[:space:]]*=/d' \
  -e '/^[[:space:]]*logging_collector[[:space:]]*=/d' \
  -e '/^[[:space:]]*log_directory[[:space:]]*=/d' \
  -e '/^[[:space:]]*log_filename[[:space:]]*=/d' \
  -e '/^[[:space:]]*log_file_mode[[:space:]]*=/d' \
  -e '/^[[:space:]]*log_rotation_age[[:space:]]*=/d' \
  -e '/^[[:space:]]*log_rotation_size[[:space:]]*=/d' \
  -e '/^[[:space:]]*log_truncate_on_rotation[[:space:]]*=/d' \
  -e '/^[[:space:]]*archive_command[[:space:]]*=/d' \
  -e '/^[[:space:]]*restore_command[[:space:]]*=/d' \
  -e '/^[[:space:]]*archive_cleanup_command[[:space:]]*=/d' \
  -e '/^[[:space:]]*lc_[a-z]*[[:space:]]*=/d' \
  -e '/^[[:space:]]*default_transaction_read_only[[:space:]]*=/d' \
  -e '/^[[:space:]]*password_encryption[[:space:]]*=/d' \
  -e '/^[[:space:]]*listen_addresses[[:space:]]*=/d' \
  "$PGDATA/postgresql.conf"

echo "Configuring stderr logging..."
echo "log_destination = 'stderr'" >> "$PGDATA/postgresql.conf"
echo "password_encryption = 'scram-sha-256'" >> "$PGDATA/postgresql.conf"
echo "logging_collector = off" >> "$PGDATA/postgresql.conf"
echo "shared_buffers = {shared_buffers_mb}MB" >> "$PGDATA/postgresql.conf"
echo "listen_addresses = '*'" >> "$PGDATA/postgresql.conf"

PG_MAJOR=$(cat "$PGDATA/PG_VERSION")

echo "Truncating postgresql.auto.conf to discard ALTER SYSTEM overrides from source..."
: > "$PGDATA/postgresql.auto.conf"

if [ ! -f "$PGDATA/pg_ident.conf" ]; then
  echo "Creating empty pg_ident.conf..."
  touch "$PGDATA/pg_ident.conf"
fi

echo "Configuring pg_hba.conf..."
cat > "$PGDATA/pg_hba.conf" << 'HBAEOF'
# TYPE  DATABASE        USER            ADDRESS                 METHOD
# trust for local: pod runs as UID 999 which has no passwd entry, so peer auth cannot resolve it
local   all             all                                     trust
host    all             all             0.0.0.0/0               scram-sha-256
host    all             all             ::/0                    scram-sha-256
HBAEOF

if [ ! -d "$PGDATA/pg_wal" ]; then
  echo "pg_wal directory missing (snapshot may be from a Windows host with WAL on a separate path), creating empty pg_wal..."
  mkdir -p "$PGDATA/pg_wal"
fi

# Run a postgres --single SQL command with a two-stage pg_resetwal -f
# fallback for snapshots whose trailing WAL is missing (e.g. taken
# mid-online-backup). For an analytics replica the priority is
# availability over byte-perfect consistency — pg_resetwal leaves the
# data dir at whatever state the snapshot captured, which is good
# enough for read-only analytics, and the alternative is a permanently
# unrecoverable replica.
#
# Stage 1: if the first attempt fails with a WAL-recovery signature,
# short-circuit straight to pg_resetwal + retry. Retrying the same
# command won't help when recovery itself is the blocker.
# Stage 2: if the first attempt fails for some other reason, try once
# more (could be transient — locked catalog, transient I/O blip).
# If the retry still fails, pg_resetwal as a last resort, then retry.
postgres_single_or_resetwal() {{
  local sql_input="$1"
  local logfile
  logfile=$(mktemp)
  set +e
  echo "$sql_input" | postgres --single -D "$PGDATA" postgres > "$logfile" 2>&1
  local rc=$?
  set -e
  cat "$logfile"
  if [ "$rc" -eq 0 ]; then
    rm -f "$logfile"
    return 0
  fi

  if grep -qE 'WAL ends before end of online backup|invalid record length at|database system was interrupted while in recovery|could not locate required checkpoint record' "$logfile"; then
    echo "WAL recovery failed (snapshot likely captured mid-online-backup) — running pg_resetwal -f and retrying" >&2
    rm -f "$logfile"
    pg_resetwal -f "$PGDATA"
    echo "$sql_input" | postgres --single -D "$PGDATA" postgres
    return $?
  fi

  echo "postgres --single failed without a recognised WAL signature — retrying once before falling back to pg_resetwal" >&2
  rm -f "$logfile"
  logfile=$(mktemp)
  set +e
  echo "$sql_input" | postgres --single -D "$PGDATA" postgres > "$logfile" 2>&1
  rc=$?
  set -e
  cat "$logfile"
  if [ "$rc" -eq 0 ]; then
    rm -f "$logfile"
    return 0
  fi

  echo "second attempt also failed — running pg_resetwal -f as a last resort and retrying" >&2
  rm -f "$logfile"
  pg_resetwal -f "$PGDATA"
  echo "$sql_input" | postgres --single -D "$PGDATA" postgres
}}

echo "Fixing database locales incompatible with this OS (single-user mode)..."
if [ "$PG_MAJOR" -ge 13 ]; then
  postgres_single_or_resetwal "UPDATE pg_database SET datcollate = 'C.UTF-8', datctype = 'C.UTF-8', datcollversion = NULL WHERE datcollate NOT IN ('C', 'C.UTF-8', 'PG_UNICODE_FAST') OR datctype NOT IN ('C', 'C.UTF-8', 'PG_UNICODE_FAST');"
else
  postgres_single_or_resetwal "UPDATE pg_database SET datcollate = 'C.UTF-8', datctype = 'C.UTF-8' WHERE datcollate NOT IN ('C', 'C.UTF-8', 'PG_UNICODE_FAST') OR datctype NOT IN ('C', 'C.UTF-8', 'PG_UNICODE_FAST');"
fi
LOCALE_CHANGED=1

echo "Starting temporary postgres to configure analytics user..."
pg_ctl -D "$PGDATA" -o "-c listen_addresses='' -c log_min_messages=WARNING" -w start

echo "Clearing restored role passwords..."
psql -U postgres -d postgres -At -c "SELECT rolname FROM pg_roles WHERE rolcanlogin AND rolname <> 'postgres' AND rolpassword IS NOT NULL" \
| while IFS= read -r role; do
  psql -U postgres -d postgres -c "ALTER ROLE \"$role\" WITH PASSWORD NULL;" 2>&1 || true
done

echo "Fixing database locales (post-startup fallback)..."
if [ "$PG_MAJOR" -ge 13 ]; then
LOCALE_CHANGED=$(psql -U postgres -d postgres -At << 'LOCALEEOF'
WITH updated AS (
  UPDATE pg_database
     SET datcollate = 'C.UTF-8', datctype = 'C.UTF-8', datcollversion = NULL
   WHERE datcollate NOT IN ('C', 'C.UTF-8', 'PG_UNICODE_FAST') OR datctype NOT IN ('C', 'C.UTF-8', 'PG_UNICODE_FAST')
  RETURNING 1
)
SELECT count(*) FROM updated;
LOCALEEOF
)
else
LOCALE_CHANGED=$(psql -U postgres -d postgres -At << 'LOCALEEOF'
WITH updated AS (
  UPDATE pg_database
     SET datcollate = 'C.UTF-8', datctype = 'C.UTF-8'
   WHERE datcollate NOT IN ('C', 'C.UTF-8', 'PG_UNICODE_FAST') OR datctype NOT IN ('C', 'C.UTF-8', 'PG_UNICODE_FAST')
  RETURNING 1
)
SELECT count(*) FROM updated;
LOCALEEOF
)
fi

for db in $(psql -U postgres -d postgres -At -c "SELECT datname FROM pg_database WHERE datallowconn AND datname <> 'template0'"); do
  echo "Fixing collations in database: $db"
  psql -U postgres -d "$db" << 'COLLEOF'
UPDATE pg_collation
   SET collcollate = 'C.UTF-8', collctype = 'C.UTF-8'
 WHERE collname = 'default';
COLLEOF
done

if [ "${{LOCALE_CHANGED:-0}}" != "0" ]; then
  echo "Locale was changed, flagging for background reindex after startup"
  touch /pgdata/needs-reindex
fi

echo "Detected PG major version: $PG_MAJOR"

# ANALYTICS_PASSWORD is generated by the operator (see replica.rs generate_password)
# and stored in a Kubernetes secret - it is not user-controlled input, so
# interpolating it directly into the SQL string is safe.
psql -U postgres -d postgres << SQLEOF
DO \$\$
BEGIN
  IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = '${{ANALYTICS_USERNAME}}') THEN
    CREATE ROLE ${{ANALYTICS_USERNAME}} WITH LOGIN PASSWORD '${{ANALYTICS_PASSWORD}}';
  ELSE
    ALTER ROLE ${{ANALYTICS_USERNAME}} WITH PASSWORD '${{ANALYTICS_PASSWORD}}';
  END IF;
END
\$\$;
SQLEOF

if [ "$PG_MAJOR" -ge 14 ] && [ "{read_only}" = "true" ]; then
  # PG >= 14 read-only: granular read role keeps the surface area minimal
  psql -U postgres -d postgres << SQLEOF
GRANT pg_read_all_data TO ${{ANALYTICS_USERNAME}};
SQLEOF
  echo "Read-only mode with PG >= 14, granted pg_read_all_data"

else
  # Read-write (any PG version) and PG < 14 read-only both go to superuser.
  # The analytics user needs DDL on existing schemas it does not own
  # (e.g. CREATE TABLE in public on PG >= 15, schema drops for
  # persistent_schemas migration, etc.), which the predefined roles
  # don't cover.
  echo "Granting superuser to analytics user..."
  psql -U postgres -d postgres << SQLEOF
ALTER ROLE ${{ANALYTICS_USERNAME}} WITH SUPERUSER;
SQLEOF
fi

if [ -f /pgdata/needs-reindex ]; then
  PGRO_STAGE=restored
else
  PGRO_STAGE=ready
fi

echo "Writing restore metadata (stage=${{PGRO_STAGE}})..."
psql -U postgres -d postgres << SQLEOF
CREATE SCHEMA IF NOT EXISTS _pgro;
CREATE TABLE IF NOT EXISTS _pgro.restore_info (
  id integer PRIMARY KEY DEFAULT 1,
  snapshot_id text NOT NULL,
  snapshot_time timestamptz,
  restored_at timestamptz NOT NULL DEFAULT now(),
  stage text NOT NULL DEFAULT 'restored',
  last_transition_time timestamptz NOT NULL DEFAULT now()
);
ALTER TABLE _pgro.restore_info ADD COLUMN IF NOT EXISTS stage text NOT NULL DEFAULT 'restored';
ALTER TABLE _pgro.restore_info ADD COLUMN IF NOT EXISTS last_transition_time timestamptz NOT NULL DEFAULT now();
INSERT INTO _pgro.restore_info (id, snapshot_id, snapshot_time, stage, last_transition_time)
VALUES (1, '${{PGRO_SNAPSHOT_ID}}', CASE WHEN '${{PGRO_SNAPSHOT_TIME}}' = '' THEN NULL ELSE '${{PGRO_SNAPSHOT_TIME}}'::timestamptz END, '${{PGRO_STAGE}}', now())
ON CONFLICT (id) DO UPDATE
  SET snapshot_id = EXCLUDED.snapshot_id,
      snapshot_time = EXCLUDED.snapshot_time,
      restored_at = now(),
      stage = EXCLUDED.stage,
      last_transition_time = now();
SQLEOF

echo "Stopping temporary postgres..."
pg_ctl -D "$PGDATA" -w stop

if [ "{read_only}" = "true" ]; then
  echo "Enabling read-only mode..."
  # Remove any existing setting to avoid duplicates across restarts
  sed -i '/^default_transaction_read_only/d' "$PGDATA/postgresql.conf"
  echo "default_transaction_read_only = on" >> "$PGDATA/postgresql.conf"
fi

echo "Auth setup complete"
"#
	);

	let labels = BTreeMap::from([
		(
			"pgro.bes.au/replica".to_string(),
			restore.spec.replica.name.clone(),
		),
		("pgro.bes.au/restore".to_string(), name.to_string()),
	]);

	let init_env = vec![
		EnvVar {
			name: "ANALYTICS_USERNAME".to_string(),
			value: Some(replica.spec.analytics_username.clone()),
			..Default::default()
		},
		env_from_secret("ANALYTICS_PASSWORD", &creds_secret, "password"),
		EnvVar {
			name: "READ_ONLY".to_string(),
			value: Some(read_only.to_string()),
			..Default::default()
		},
		EnvVar {
			name: "PGRO_SNAPSHOT_ID".to_string(),
			value: Some(restore.spec.snapshot.clone()),
			..Default::default()
		},
		EnvVar {
			name: "PGRO_SNAPSHOT_TIME".to_string(),
			value: Some(restore.spec.snapshot_time.clone().unwrap_or_default()),
			..Default::default()
		},
	];

	Ok(Deployment {
		metadata: ObjectMeta {
			name: Some(name.to_string()),
			namespace: Some(namespace.to_string()),
			labels: Some(labels.clone()),
			owner_references: Some(vec![restore_owner_reference(restore)]),
			..Default::default()
		},
		spec: Some(DeploymentSpec {
			replicas: Some(1),
			selector: LabelSelector {
				match_labels: Some(BTreeMap::from([(
					"pgro.bes.au/restore".to_string(),
					name.to_string(),
				)])),
				..Default::default()
			},
			progress_deadline_seconds: Some(3600),
			strategy: Some(k8s_openapi::api::apps::v1::DeploymentStrategy {
				type_: Some("Recreate".to_string()),
				..Default::default()
			}),
			template: PodTemplateSpec {
				metadata: Some(ObjectMeta {
					labels: Some(labels),
					annotations: replica.spec.pod_annotations.clone(),
					..Default::default()
				}),
				spec: Some(PodSpec {
					security_context: Some(k8s_openapi::api::core::v1::PodSecurityContext {
						run_as_user: Some(999),
						run_as_group: Some(999),
						fs_group: Some(999),
						..Default::default()
					}),
					init_containers: Some(vec![
						Container {
							name: "fix-locale".to_string(),
							image: Some(pg_image.clone()),
							command: Some(vec!["/bin/sh".to_string(), "-c".to_string()]),
							args: Some(vec![locale_script]),
							security_context: Some(
								k8s_openapi::api::core::v1::SecurityContext {
									run_as_user: Some(0),
									run_as_group: Some(0),
									..Default::default()
								},
							),
							volume_mounts: Some(vec![VolumeMount {
								name: "locale-data".to_string(),
								mount_path: "/locale-data".to_string(),
								..Default::default()
							}]),
							resources: Some(ResourceRequirements {
								requests: Some(BTreeMap::from([
									("cpu".to_string(), Quantity("50m".to_string())),
									("memory".to_string(), Quantity("64Mi".to_string())),
								])),
								..Default::default()
							}),
							..Default::default()
						},
						Container {
							name: "setup-auth".to_string(),
							image: Some(pg_image.clone()),
							command: Some(vec!["/bin/sh".to_string(), "-c".to_string()]),
							args: Some(vec![init_script]),
							env: Some(init_env),
							volume_mounts: Some(vec![
								VolumeMount {
									name: "pgdata".to_string(),
									mount_path: "/pgdata".to_string(),
									..Default::default()
								},
								VolumeMount {
									name: "locale-data".to_string(),
									mount_path: "/usr/lib/locale".to_string(),
									..Default::default()
								},
							]),
							resources: Some(ResourceRequirements {
								requests: Some(BTreeMap::from([
									("cpu".to_string(), Quantity("100m".to_string())),
									("memory".to_string(), Quantity("128Mi".to_string())),
								])),
								limits: Some(BTreeMap::from([
									("cpu".to_string(), Quantity("500m".to_string())),
									("memory".to_string(), Quantity("256Mi".to_string())),
								])),
								..Default::default()
							}),
							..Default::default()
						},
					]),
					containers: vec![Container {
						name: "postgres".to_string(),
						image: Some(pg_image),
						command: Some(vec!["/bin/sh".to_string(), "-c".to_string()]),
						args: Some(vec![r#"
if [ -f /pgdata/needs-reindex ]; then
  PG_MAJOR=$(cat /pgdata/pgdata/PG_VERSION)
  (
    while ! pg_isready -q -U postgres -d postgres; do sleep 2; done
    psql -U postgres -d postgres -c "UPDATE _pgro.restore_info SET stage = 'reindexing', last_transition_time = now() WHERE id = 1;"
    for db in $(psql -U postgres -d postgres -At -c "SELECT datname FROM pg_database WHERE datallowconn AND datname <> 'template0'"); do
      INDEXES=$(psql -U postgres -d "$db" -At -c "
        SELECT DISTINCT indexrelid::regclass::text
        FROM pg_index i
        JOIN pg_attribute a ON a.attrelid = i.indexrelid
        WHERE a.attcollation <> 0 AND i.indisvalid;
      ")
      COUNT=$(echo "$INDEXES" | grep -c . || true)
      echo "Reindex after locale change: $db ($COUNT collation-dependent indexes)"
      N=0
      echo "$INDEXES" | while IFS= read -r idx; do
        [ -z "$idx" ] && continue
        N=$((N + 1))
        echo "  [$N/$COUNT] $db: $idx"
        if [ "$PG_MAJOR" -ge 14 ]; then
          psql -U postgres -d "$db" -c "REINDEX INDEX CONCURRENTLY $idx;" 2>&1 || true
        else
          psql -U postgres -d "$db" -c "REINDEX INDEX $idx;" 2>&1 || true
        fi
      done
    done
    rm -f /pgdata/needs-reindex
    psql -U postgres -d postgres -c "UPDATE _pgro.restore_info SET stage = 'ready', last_transition_time = now() WHERE id = 1;"
    echo "Background reindex complete"
  ) &
fi
exec postgres -D /pgdata/pgdata ${PGRO_LOG_LEVEL:+-c log_min_messages=$PGRO_LOG_LEVEL}
"#.to_string()]),
						env: Some(vec![
							EnvVar {
								name: "PGDATA".to_string(),
								value: Some("/pgdata/pgdata".to_string()),
								..Default::default()
							},
							EnvVar {
								name: "POSTGRES_HOST_AUTH_METHOD".to_string(),
								value: Some("scram-sha-256".to_string()),
								..Default::default()
							},
						]),
						ports: Some(vec![ContainerPort {
							name: Some("postgres".to_string()),
							container_port: 5432,
							protocol: Some("TCP".to_string()),
							..Default::default()
						}]),
						volume_mounts: Some(vec![
							VolumeMount {
								name: "pgdata".to_string(),
								mount_path: "/pgdata".to_string(),
								..Default::default()
							},
							VolumeMount {
								name: "locale-data".to_string(),
								mount_path: "/usr/lib/locale".to_string(),
								..Default::default()
							},
							VolumeMount {
								name: "dshm".to_string(),
								mount_path: "/dev/shm".to_string(),
								..Default::default()
							},
						]),
						readiness_probe: Some(Probe {
							exec: Some(ExecAction {
								command: Some(vec![
									"/bin/sh".to_string(),
									"-c".to_string(),
									"pg_isready -U postgres -d postgres && [ ! -f /pgdata/needs-reindex ]".to_string(),
								]),
							}),
							initial_delay_seconds: Some(5),
							period_seconds: Some(5),
							timeout_seconds: Some(3),
							failure_threshold: Some(6),
							..Default::default()
						}),
						liveness_probe: Some(Probe {
							exec: Some(ExecAction {
								command: Some(vec![
									"pg_isready".to_string(),
									"-U".to_string(),
									"postgres".to_string(),
									"-d".to_string(),
									"postgres".to_string(),
								]),
							}),
							initial_delay_seconds: Some(30),
							period_seconds: Some(10),
							timeout_seconds: Some(3),
							failure_threshold: Some(3),
							..Default::default()
						}),
						resources: replica.spec.resources.clone(),
						..Default::default()
					}],
					volumes: Some(vec![
						Volume {
							name: "pgdata".to_string(),
							persistent_volume_claim: Some(
								k8s_openapi::api::core::v1::PersistentVolumeClaimVolumeSource {
									claim_name: pvc_name,
									read_only: Some(false),
								},
							),
							..Default::default()
						},
						Volume {
							name: "locale-data".to_string(),
							empty_dir: Some(
								k8s_openapi::api::core::v1::EmptyDirVolumeSource::default(),
							),
							..Default::default()
						},
						Volume {
							name: "dshm".to_string(),
							empty_dir: Some(
								k8s_openapi::api::core::v1::EmptyDirVolumeSource {
										medium: Some("Memory".to_string()),
										size_limit: Some(shm_size),
									},
							),
							..Default::default()
						},
					]),
					affinity: replica.spec.affinity.clone(),
					tolerations: Some(replica.spec.tolerations.clone()),
					..Default::default()
				}),
			},
			..Default::default()
		}),
		..Default::default()
	})
}