// SF Extension — Persistent Notification Store
// Captures durable ctx.ui.notify() calls and workflow-logger errors to
// .sf/notifications.jsonl so they survive context resets and session restarts.
// Rotates at MAX_ENTRIES to prevent unbounded growth.
//
// Long-term hygiene (schema v2+): repeated equivalent notices merge into one row
// with repeatCount + lastTs; blocking/action-required rows never merge.
// Pass metadata.dedupe_key for stable identity; metadata.noticeKind for classification.
import { randomUUID } from "node:crypto";
import {
	appendFileSync,
	closeSync,
	existsSync,
	mkdirSync,
	openSync,
	readFileSync,
	renameSync,
	unlinkSync,
	writeFileSync,
} from "node:fs";
import { join } from "node:path";
import { sfRuntimeRoot } from "./paths.js";

// ─── Constants ──────────────────────────────────────────────────────────
const MAX_ENTRIES = 500;
const FILENAME = "notifications.jsonl";
const LOCKFILE = "notifications.lock";
/** Legacy rows on disk may omit schemaVersion — treated as v1 */
export const NOTIFICATION_SCHEMA_VERSION_MIN = 1;
/** Current write schema — adds repeatCount, lastTs, noticeKind */
export const NOTIFICATION_SCHEMA_VERSION_WRITE = 2;
const DURABLE_DEDUP_WINDOW_MS = 60 * 60 * 1000;
const DURABLE_DEDUP_SCAN_LIMIT = 100;
const ACTIONABLE_KINDS = new Set([
	"action_required",
	"approval_request",
	"attention",
	"blocker",
	"error",
	"terminal",
]);
const NOISY_STATUS_PATTERNS = [
	/^MCP client ready\b/,
	/^Startup doctor: applied\b/,
	/^Flow audit: Inspect session\b/,
	/^UOK parity report shows\b/,
	/Permission bypassed - all checks disabled!?$/,
	/^Dynamic routing:/,
	/^Pre-flight:/,
	/^\[unit\]/,
	/^Model \[/,
	/^\[mcp\]/,
	/^Assisted mode started\b/,
	/^Autonomous mode resumed\b/,
	/^Resuming paused session\b/,
	/^Resumed paused session\b/,
];

/**
 * Optional classification for automated vs human-visible notices (see docs/product-specs/notification-source-hygiene.md).
 *
 * Purpose: let renderers group and style notices without parsing message text.
 *
 * Consumer: ctx.ui.notify metadata, headless classifiers, notification overlay.
 */
export const NOTICE_KIND = {
	SYSTEM_NOTICE: "system_notice",
	TOOL_NOTICE: "tool_notice",
	BLOCKING_NOTICE: "blocking_notice",
	USER_VISIBLE: "user_visible",
};

// ─── Module State ───────────────────────────────────────────────────────
let _basePath = null;
let _lineCount = 0; // Hint for rotation — not authoritative for public API
let _suppressCount = 0;
const _changeListeners = new Set();
/** Count of appendNotification failures since last reset — surfaced by status checks. */
let _appendFailureCount = 0;
/** Last append failure details for diagnostics. */
let _lastAppendFailure = null;
// ─── Public API ─────────────────────────────────────────────────────────
/**
 * Initialize the notification store. Call once at session start with the
 * project root. Seeds in-memory counters from the existing file on disk.
 */
export function initNotificationStore(basePath) {
	_basePath = basePath;
	// Seed line count hint for rotation — public counters read from disk
	_lineCount = _readEntriesFromDisk(basePath).length;
}
/**
 * Append a notification entry to the store. Synchronous — safe to call
 * from the notify() shim which is declared void (not async).
 *
 * Duplicate notices (same dedupe key within the durable window, unread): merge
 * into the existing row (repeatCount++, lastTs updated) instead of growing the log.
 * Set metadata.merge === false to force a new row. Blocking / approval / blocker
 * kinds do not merge — each emission stays distinct for consent surfaces.
 */
export function appendNotification(
	message,
	severity,
	source = "notify",
	metadata,
) {
	if (!_basePath) return;
	if (_suppressCount > 0) return;
	const normalizedSeverity = severity === "warn" ? "warning" : severity;
	const persistedMessage =
		message.length > 500 ? message.slice(0, 500) + "…" : message;
	if (
		!shouldPersistNotification(normalizedSeverity, metadata, persistedMessage)
	)
		return;
	const now = Date.now();
	if (
		tryMergePersistedNotification(_basePath, {
			normalizedSeverity,
			source,
			persistedMessage,
			metadata,
			now,
		})
	) {
		_emitChange();
		return;
	}
	const entry = {
		schemaVersion: NOTIFICATION_SCHEMA_VERSION_WRITE,
		id: randomUUID(),
		ts: new Date(now).toISOString(),
		severity: normalizedSeverity,
		message: persistedMessage,
		source,
		read: false,
		repeatCount: 1,
		...(metadata?.noticeKind ? { noticeKind: metadata.noticeKind } : {}),
		...(metadata ? { metadata } : {}),
	};
	try {
		const dir = sfRuntimeRoot(_basePath);
		mkdirSync(dir, { recursive: true });
		appendFileSync(join(dir, FILENAME), JSON.stringify(entry) + "\n", "utf-8");
		_lineCount++;
		// Rotate if hint suggests we're over limit
		if (_lineCount > MAX_ENTRIES) {
			_rotate();
		}
		_emitChange();
	} catch (err) {
		// Non-fatal — never let persistence break the caller — but track failure
		// so callers can distinguish persisted notifications from failed ones.
		_appendFailureCount++;
		_lastAppendFailure = {
			ts: new Date().toISOString(),
			correlationId: entry.id,
			error: err?.message ?? String(err),
		};
		_writeStderr(
			`[sf:notification-store] append failed for ${entry.id}: ${err?.message ?? String(err)}\n`,
		);
	}
}
function shouldPersistNotification(severity, metadata, message) {
	if (metadata?.kind === "progress") return false;
	if (metadata?.persist === false) return false;
	if (metadata?.persist === true) return true;
	if (metadata?.blocking === true || metadata?.requires_action === true) {
		return true;
	}
	if (ACTIONABLE_KINDS.has(metadata?.kind)) return true;
	if (severity === "error") return true;
	if (isNoisyStatusMessage(message)) return false;
	return severity === "warning";
}
/**
 * Read all notification entries from disk. Returns newest-first.
 */
export function readNotifications(basePath) {
	const bp = basePath ?? _basePath;
	if (!bp) return [];
	return _readEntriesFromDisk(bp).reverse();
}
/**
 * Mark all notifications as read. Atomic rewrite via temp-file + rename.
 * Resyncs in-memory counters from disk after mutation.
 */
export function markAllRead(basePath) {
	const bp = basePath ?? _basePath;
	if (!bp) return;
	const entries = _readEntriesFromDisk(bp);
	if (entries.length === 0) return;
	const hasUnread = entries.some((e) => !e.read);
	if (!hasUnread) return;
	let changed = false;
	try {
		_withLock(bp, () => {
			// Re-read inside lock to get freshest state
			const fresh = _readEntriesFromDisk(bp);
			if (fresh.length === 0 || !fresh.some((e) => !e.read)) return;
			const lines = fresh.map((e) => JSON.stringify({ ...e, read: true }));
			_atomicWrite(bp, lines.join("\n") + "\n");
			changed = true;
		});
	} catch {
		// Non-fatal
	}
	if (changed) _emitChange();
}
/**
 * Clear all notifications. Atomic write of empty content under lock.
 */
export function clearNotifications(basePath) {
	const bp = basePath ?? _basePath;
	if (!bp) return;
	try {
		_withLock(bp, () => {
			_atomicWrite(bp, "");
		});
		_lineCount = 0;
		_emitChange();
	} catch {
		// Non-fatal
	}
}
/**
 * Get the current unread count. Reads from disk to stay accurate across
 * processes (web subprocess can clear/modify the file independently).
 */
export function getUnreadCount() {
	if (!_basePath) return 0;
	try {
		const entries = _readEntriesFromDisk(_basePath);
		return entries.filter((e) => !e.read).length;
	} catch {
		return 0;
	}
}
/**
 * Get the total notification count. Reads from disk for cross-process accuracy.
 */
export function getLineCount() {
	if (!_basePath) return 0;
	try {
		return _readEntriesFromDisk(_basePath).length;
	} catch {
		return 0;
	}
}
/**
 * Temporarily suppress persistence. Use around ctx.ui.notify calls that
 * should NOT be persisted (e.g., confirmation toasts after clear).
 * Calls are ref-counted — nest safely.
 */
export function suppressPersistence() {
	_suppressCount++;
}
export function unsuppressPersistence() {
	_suppressCount = Math.max(0, _suppressCount - 1);
}
/**
 * Subscribe to notification-store mutations (append, mark-read, clear).
 * Returns an unsubscribe function.
 */
export function onNotificationStoreChange(listener) {
	_changeListeners.add(listener);
	return () => {
		_changeListeners.delete(listener);
	};
}

/**
 * Return the count of appendNotification failures since last reset.
 * Consumed by doctor/status to surface persistent notification-store divergence.
 */
export function getAppendFailureCount() {
	return _appendFailureCount;
}

/**
 * Return the last append failure details, or null if none since reset.
 */
export function getLastAppendFailure() {
	return _lastAppendFailure;
}

// ─── Test Helpers ───────────────────────────────────────────────────────
/**
 * Reset module state. Only for use in tests.
 */
export function _resetNotificationStore() {
	_basePath = null;
	_lineCount = 0;
	_suppressCount = 0;
	_changeListeners.clear();
	_appendFailureCount = 0;
	_lastAppendFailure = null;
}
// ─── Internal ───────────────────────────────────────────────────────────
function _readEntriesFromDisk(basePath) {
	const filePath = join(sfRuntimeRoot(basePath), FILENAME);
	if (!existsSync(filePath)) return [];
	try {
		const content = readFileSync(filePath, "utf-8");
		return content
			.split("\n")
			.filter((l) => l.length > 0)
			.map((l) => {
				try {
					return normalizeNotificationEntry(JSON.parse(l));
				} catch {
					return null;
				}
			})
			.filter((e) => e !== null);
	} catch {
		return [];
	}
}
function normalizeNotificationEntry(entry) {
	if (!entry || typeof entry !== "object" || Array.isArray(entry)) return null;
	const rawSv = entry.schemaVersion;
	const schemaVersion =
		rawSv === undefined || rawSv === null
			? NOTIFICATION_SCHEMA_VERSION_MIN
			: rawSv;
	if (
		schemaVersion < NOTIFICATION_SCHEMA_VERSION_MIN ||
		schemaVersion > NOTIFICATION_SCHEMA_VERSION_WRITE
	) {
		return null;
	}
	return {
		...entry,
		schemaVersion,
		read: entry.read === true,
		repeatCount: entry.repeatCount ?? 1,
	};
}
/**
 * Whether two emissions with the same normalized key should collapse into one row.
 * Distinct consent surfaces (approval/blocker kinds) never merge; blocking:true alone
 * still merges when dedupe_key matches so operators see one row with repeatCount.
 */
function shouldMergeDuplicates(metadata) {
	if (metadata?.merge === false) return false;
	const k = metadata?.kind;
	if (k === "blocker" || k === "approval_request" || k === "action_required") {
		return false;
	}
	return true;
}
/**
 * Scan recent unread rows for the same logical key; merge repeatCount + lastTs.
 *
 * Purpose: durable grouping per notification-source-hygiene — fewer duplicate lines,
 * preserved counts for operators.
 *
 * Consumer: appendNotification only.
 */
function tryMergePersistedNotification(basePath, params) {
	const { normalizedSeverity, source, persistedMessage, metadata, now } =
		params;
	if (!shouldMergeDuplicates(metadata)) return false;

	const keySeed =
		metadata?.dedupe_key ??
		`${normalizedSeverity}:${source}:${persistedMessage}`;
	const normalizedKey = normalizeDedupKey(keySeed);

	let merged = false;
	try {
		_withLock(basePath, () => {
			const entries = _readEntriesFromDisk(basePath);
			const scanStart = Math.max(0, entries.length - DURABLE_DEDUP_SCAN_LIMIT);
			for (let i = entries.length - 1; i >= scanStart; i--) {
				const entry = entries[i];
				if (entry.read === true) continue;
				const ts = Date.parse(entry.ts);
				if (!Number.isFinite(ts) || now - ts > DURABLE_DEDUP_WINDOW_MS)
					continue;
				const entryKey = entry.metadata?.dedupe_key
					? normalizeDedupKey(entry.metadata.dedupe_key)
					: normalizeDedupKey(
							`${entry.severity}:${entry.source ?? "notify"}:${entry.message}`,
						);
				if (entryKey !== normalizedKey) continue;

				const nextRepeat = (entry.repeatCount ?? 1) + 1;
				const mergedEntry = {
					...entry,
					schemaVersion: Math.max(
						entry.schemaVersion ?? NOTIFICATION_SCHEMA_VERSION_MIN,
						NOTIFICATION_SCHEMA_VERSION_WRITE,
					),
					repeatCount: nextRepeat,
					lastTs: new Date(now).toISOString(),
				};
				if (metadata?.noticeKind && !entry.noticeKind) {
					mergedEntry.noticeKind = metadata.noticeKind;
				}
				entries[i] = mergedEntry;
				_atomicWrite(
					basePath,
					entries.map((e) => JSON.stringify(e)).join("\n") + "\n",
				);
				_lineCount = entries.length;
				merged = true;
				return;
			}
		});
	} catch {
		return false;
	}
	return merged;
}
function normalizeDedupKey(value) {
	return String(value)
		.replace(/\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z/g, "<ts>")
		.replace(/\bsince\s+[^.]+/gi, "since <elapsed>")
		.trim();
}
function isNoisyStatusMessage(message) {
	return NOISY_STATUS_PATTERNS.some((pattern) => pattern.test(message));
}
function _rotate() {
	if (!_basePath) return;
	try {
		_withLock(_basePath, () => {
			// Re-read inside lock to get freshest state
			const entries = _readEntriesFromDisk(_basePath);
			if (entries.length <= MAX_ENTRIES) return;
			const trimmed = entries.slice(entries.length - MAX_ENTRIES);
			const lines = trimmed.map((e) => JSON.stringify(e));
			_atomicWrite(_basePath, lines.join("\n") + "\n");
			_lineCount = trimmed.length;
		});
	} catch {
		// Non-fatal
	}
}
function _emitChange() {
	for (const listener of _changeListeners) {
		try {
			listener();
		} catch {
			// Non-fatal
		}
	}
}
/**
 * Atomic file rewrite via temp-file + rename. Prevents partial reads
 * by other processes (web API subprocess, parallel workers).
 * Must be called inside _withLock for cross-process safety.
 */
function _atomicWrite(basePath, content) {
	const dir = sfRuntimeRoot(basePath);
	mkdirSync(dir, { recursive: true });
	const target = join(dir, FILENAME);
	const tmp = target + ".tmp." + process.pid;
	writeFileSync(tmp, content, "utf-8");
	renameSync(tmp, target);
}
/**
 * Acquire an exclusive lockfile for rewrite operations.
 * Uses O_CREAT|O_EXCL for atomic creation — if the file exists, another
 * process holds the lock. Retries briefly, then proceeds anyway (best-effort)
 * to avoid deadlocking the UI on a stale lock.
 */
function _withLock(basePath, fn) {
	const runtimeDir = sfRuntimeRoot(basePath);
	const lockPath = join(runtimeDir, LOCKFILE);
	let fd = null;
	const maxAttempts = 5;
	const retryMs = 20;
	for (let i = 0; i < maxAttempts; i++) {
		try {
			mkdirSync(runtimeDir, { recursive: true });
			fd = openSync(lockPath, "wx");
			break;
		} catch (err) {
			if (err?.code === "EEXIST") {
				// Check if lock is stale (older than 5s)
				try {
					const stat = readFileSync(lockPath, "utf-8");
					const lockTime = parseInt(stat, 10);
					// Treat NaN (creator crashed before writing timestamp) as stale.
					if (
						Number.isNaN(lockTime) ||
						(Number.isFinite(lockTime) && Date.now() - lockTime > 5000)
					) {
						try {
							unlinkSync(lockPath);
						} catch {
							/* race ok */
						}
						continue;
					}
				} catch {
					/* can't read lock, retry */
				}
				// Wait and retry
				const start = Date.now();
				while (Date.now() - start < retryMs) {
					/* spin */
				}
				continue;
			}
			// Other error — proceed without lock
			break;
		}
	}
	// Best-effort: mutation runs regardless of lock status (idempotent overwrites).
	// createdLock gates cleanup only — never skip fn() on lock failure.
	const createdLock = fd !== null;
	try {
		if (createdLock && fd !== null) {
			// Write our PID timestamp into the lock for stale detection
			writeFileSync(lockPath, String(Date.now()), "utf-8");
			closeSync(fd);
		}
		return fn();
	} finally {
		// Only delete the lock if we created it — never remove another process's lock
		if (createdLock) {
			try {
				unlinkSync(lockPath);
			} catch {
				/* best-effort cleanup */
			}
		}
	}
}

function _writeStderr(message) {
	process.stderr.write(message);
}
