// Orchestrates one backup cycle: enumerate targets, then for each account
// snapshot the repo + sync new blobs, respecting account status.

import { and, eq } from 'drizzle-orm';
import db from '../database';
import {
	account,
	backup_run,
	blob_backup,
	repo_backup
} from '../database/schema';
import type { AccountStatus } from '../database/schema';
import {
	describeRepo,
	getRepoStatus,
	iterateBlobs,
	streamBlob,
	streamRepo,
	StreamXrpcError
} from '../atproto/client';
import { storage } from '../storage';
import { collectTargets, type BackupTarget } from './targets';
import { pruneAccountBackups } from './retention';

// Account is "healthy" — i.e. we should keep backing it up and pruning old
// snapshots. Non-active statuses freeze the archive in place.
export function isHealthy(status: AccountStatus): boolean {
	return status === 'active';
}

// ISO-ish timestamp safe for filenames (no colons).
function fsTimestamp(d: Date): string {
	return d
		.toISOString()
		.replace(/\.\d{3}Z$/, 'Z')
		.replace(/:/g, '-');
}

interface RunError {
	did: string;
	stage: string;
	message: string;
}

export interface RunSummary {
	accountsConsidered: number;
	accountsBackedUp: number;
	repoSnapshotsTaken: number;
	blobsFetched: number;
	bytesWritten: number;
	errors: RunError[];
}

async function backupOneAccount(
	target: BackupTarget,
	errors: RunError[]
): Promise<{
	tookSnapshot: boolean;
	blobsFetched: number;
	bytesWritten: number;
}> {
	const out = { tookSnapshot: false, blobsFetched: 0, bytesWritten: 0 };
	const now = new Date();
	const s = storage();

	// --- Confirm status via getRepoStatus (authoritative for this cycle). ---
	let status: AccountStatus = target.initial_status;
	let currentRev = target.latest_rev ?? undefined;
	try {
		const st = await getRepoStatus(target.pds_endpoint, target.did);
		if (st.active === false) {
			const knownStatuses: AccountStatus[] = [
				'takendown',
				'suspended',
				'deleted',
				'deactivated',
				'desynchronized',
				'throttled'
			];
			status = (knownStatuses as string[]).includes(st.status ?? '')
				? (st.status as AccountStatus)
				: 'unknown';
		} else {
			status = 'active';
		}
		currentRev = st.rev ?? currentRev;
	} catch (err) {
		errors.push({
			did: target.did,
			stage: 'getRepoStatus',
			message: (err as Error).message
		});
		// Fall through with whatever we had from listRepos.
	}

	await db
		.update(account)
		.set({
			status,
			last_status_checked_at: now
		})
		.where(eq(account.did, target.did));

	// --- If not healthy, stop here. No new snapshots, no pruning. ---
	if (!isHealthy(status)) return out;

	const existing = await db.query.account.findFirst({
		where: eq(account.did, target.did)
	});

	// --- Refresh handle every cycle. listRepos doesn't return it, and handles
	// are mutable (domain transfers, re-points), so we can't rely on the
	// stored value being current. describeRepo on the same PDS is cheap and
	// authoritative. On failure we leave the previous value in place.
	try {
		const desc = await describeRepo(target.pds_endpoint, target.did);
		if (desc.handle && desc.handle !== existing?.handle) {
			await db
				.update(account)
				.set({ handle: desc.handle })
				.where(eq(account.did, target.did));
		}
	} catch {
		// non-fatal; retry next cycle.
	}

	const previousRev = existing?.last_archived_rev;

	// --- Repo snapshot ---
	if (currentRev && currentRev === previousRev) {
		// No-op: the PDS reports the same rev we already archived.
	} else {
		try {
			const stampedRev = currentRev ?? fsTimestamp(now);
			const storagePath = `repos/${target.did}/${fsTimestamp(now)}_${stampedRev}.car`;
			const res = await streamRepo(target.pds_endpoint, target.did);
			if (!res.body) throw new Error('getRepo returned no body');
			await s.put(storagePath, res.body);
			const stat = await s.stat(storagePath);
			const size = stat?.size ?? 0;

			await db
				.insert(repo_backup)
				.values({
					account_did: target.did,
					rev: stampedRev,
					storage_path: storagePath,
					size_bytes: size,
					taken_at: now
				})
				.onConflictDoNothing();

			await db
				.update(account)
				.set({
					last_archived_rev: stampedRev,
					last_repo_backup_at: now
				})
				.where(eq(account.did, target.did));

			out.tookSnapshot = true;
			out.bytesWritten += size;
		} catch (err) {
			errors.push({
				did: target.did,
				stage: 'getRepo',
				message:
					err instanceof StreamXrpcError
						? `${err.status} ${err.body.slice(0, 120)}`
						: (err as Error).message
			});
		}
	}

	// --- Blob sync (incremental by CID) ---
	try {
		// Load the set of CIDs we already have for this account. Blob backups
		// are forever-kept, so anything in the table is durable.
		const existingRows = await db
			.select({ cid: blob_backup.cid })
			.from(blob_backup)
			.where(eq(blob_backup.account_did, target.did));
		const existing = new Set(existingRows.map((r) => r.cid));

		for await (const cid of iterateBlobs(target.pds_endpoint, target.did)) {
			if (existing.has(cid)) continue;
			try {
				const res = await streamBlob(target.pds_endpoint, target.did, cid);
				if (!res.body) throw new Error('getBlob returned no body');
				const storagePath = `blobs/${target.did}/${cid}`;
				await s.put(storagePath, res.body);
				const stat = await s.stat(storagePath);
				const size = stat?.size ?? 0;
				await db
					.insert(blob_backup)
					.values({
						account_did: target.did,
						cid,
						storage_path: storagePath,
						size_bytes: size
					})
					.onConflictDoNothing();
				existing.add(cid);
				out.blobsFetched += 1;
				out.bytesWritten += size;
			} catch (err) {
				errors.push({
					did: target.did,
					stage: 'getBlob',
					message: `${cid}: ${(err as Error).message}`
				});
			}
		}
		await db
			.update(account)
			.set({ last_blob_sync_at: now })
			.where(eq(account.did, target.did));
	} catch (err) {
		errors.push({
			did: target.did,
			stage: 'listBlobs',
			message: (err as Error).message
		});
	}

	// --- Retention pruning (only for healthy accounts). ---
	try {
		await pruneAccountBackups(target.did, now);
	} catch (err) {
		errors.push({
			did: target.did,
			stage: 'prune',
			message: (err as Error).message
		});
	}

	return out;
}

export async function runBackupCycle(): Promise<RunSummary> {
	const startedAt = new Date();
	const [inserted] = await db
		.insert(backup_run)
		.values({ started_at: startedAt })
		.returning({ id: backup_run.id });
	const runId = inserted!.id;

	const { targets, pdsErrors } = await collectTargets();
	const errors: RunError[] = pdsErrors.map((e) => ({
		did: '(pds)',
		stage: 'listRepos',
		message: `${e.hostname}: ${e.message}`
	}));

	let accountsBackedUp = 0;
	let repoSnapshotsTaken = 0;
	let blobsFetched = 0;
	let bytesWritten = 0;

	for (const t of targets) {
		try {
			const res = await backupOneAccount(t, errors);
			if (res.tookSnapshot || res.blobsFetched > 0) accountsBackedUp += 1;
			if (res.tookSnapshot) repoSnapshotsTaken += 1;
			blobsFetched += res.blobsFetched;
			bytesWritten += res.bytesWritten;
		} catch (err) {
			errors.push({
				did: t.did,
				stage: 'account',
				message: (err as Error).message
			});
		}
	}

	const summary: RunSummary = {
		accountsConsidered: targets.length,
		accountsBackedUp,
		repoSnapshotsTaken,
		blobsFetched,
		bytesWritten,
		errors
	};

	await db
		.update(backup_run)
		.set({
			finished_at: new Date(),
			accounts_considered: summary.accountsConsidered,
			accounts_backed_up: summary.accountsBackedUp,
			repo_snapshots_taken: summary.repoSnapshotsTaken,
			blobs_fetched: summary.blobsFetched,
			bytes_written: summary.bytesWritten,
			errors: errors.length ? JSON.stringify(errors) : null
		})
		.where(eq(backup_run.id, runId));

	return summary;
}