import { ensureMessageInstances } from "../message-coercion.js";
import { MessageAssembler } from "../../client/stream/messages.js";
import { isRootNamespace, namespaceKey } from "../namespace.js";
import { assembledMessageToBaseMessage } from "../assembled-to-message.js";
import { buildMessageIndex, reconcileMessagesFromValues, shouldPreferValuesMessageForToolCalls } from "../message-reconciliation.js";
import { openProjectionSubscription } from "./runtime.js";
//#region src/stream/projections/messages.ts
function messagesProjection(namespace) {
	const ns = [...namespace];
	return {
		key: `messages|${namespaceKey(ns)}`,
		namespace: ns,
		initial: [],
		open({ thread, store, rootBus }) {
			const assembler = new MessageAssembler();
			const roleByKey = /* @__PURE__ */ new Map();
			const indexById = /* @__PURE__ */ new Map();
			const streamMessageIds = /* @__PURE__ */ new Set();
			let valuesMessageIds = /* @__PURE__ */ new Set();
			if (isRootNamespace(ns) && rootBus.channels.includes("messages")) {
				const unsubscribe = rootBus.subscribe((event) => {
					if (event.method !== "messages") return;
					if (!isRootNamespace(event.params.namespace)) return;
					applyEvent(event);
				});
				return { dispose() {
					unsubscribe();
				} };
			}
			let disposed = false;
			const pendingMessages = [];
			let dirty = false;
			let flushScheduled = false;
			const flushChannel = typeof MessageChannel !== "undefined" ? new MessageChannel() : null;
			const flush = () => {
				flushScheduled = false;
				if (!dirty || disposed) return;
				dirty = false;
				store.setValue(pendingMessages.slice());
			};
			if (flushChannel != null) flushChannel.port1.onmessage = flush;
			const scheduleFlush = () => {
				dirty = true;
				if (flushScheduled) return;
				flushScheduled = true;
				if (flushChannel != null) flushChannel.port2.postMessage(null);
				else setTimeout(flush, 0);
			};
			const applyValuesEvent = (event) => {
				const data = event.params.data;
				if (data == null || typeof data !== "object" || Array.isArray(data)) return;
				const rawMessages = data.messages;
				if (!Array.isArray(rawMessages) || rawMessages.length === 0) return;
				const reconciliation = reconcileMessagesFromValues({
					valueMessages: ensureMessageInstances(rawMessages),
					currentMessages: pendingMessages,
					currentIndexById: indexById,
					previousValueMessageIds: valuesMessageIds,
					streamedMessageIds: streamMessageIds,
					preferValuesMessage: shouldPreferValuesMessageForToolCalls
				});
				valuesMessageIds = reconciliation.valueMessageIds;
				const reconciledMessages = [...reconciliation.messages];
				pendingMessages.length = 0;
				for (const message of reconciledMessages) pendingMessages.push(message);
				indexById.clear();
				for (const [id, idx] of buildMessageIndex(pendingMessages)) indexById.set(id, idx);
				scheduleFlush();
			};
			const applyEvent = (event) => {
				const data = event.params.data;
				if (data.event === "message-start") {
					const startData = data;
					const role = startData.role ?? "ai";
					const extendedRole = startData.role ?? role;
					const maybeToolCallId = startData.tool_call_id;
					if (startData.id != null) roleByKey.set(startData.id, {
						role: extendedRole,
						toolCallId: maybeToolCallId
					});
				}
				const update = assembler.consume(event);
				if (update == null) return;
				const msg = update.message;
				const id = msg.id;
				if (id == null) return;
				const captured = roleByKey.get(id) ?? { role: "ai" };
				const base = assembledMessageToBaseMessage(msg, captured.role, { toolCallId: captured.toolCallId });
				streamMessageIds.add(id);
				const existingIdx = indexById.get(id);
				if (existingIdx == null) {
					indexById.set(id, pendingMessages.length);
					pendingMessages.push(base);
				} else pendingMessages[existingIdx] = base;
				scheduleFlush();
			};
			let runtime;
			const openSubscription = () => {
				runtime = openProjectionSubscription({
					thread,
					channels: ["messages", "values"],
					namespace: ns,
					onEvent(event) {
						if (event.method === "messages") applyEvent(event);
						else if (event.method === "values") applyValuesEvent(event);
					}
				});
			};
			(async () => {
				if (!(await rootBus.trySeedFromHistory?.({
					kind: "messages",
					namespace: ns,
					store
				}) === true) && !disposed) openSubscription();
			})();
			return { async dispose() {
				disposed = true;
				if (flushChannel != null) {
					flushChannel.port1.onmessage = null;
					flushChannel.port1.close();
					flushChannel.port2.close();
				}
				await runtime?.dispose();
			} };
		}
	};
}
//#endregion
export { messagesProjection };

//# sourceMappingURL=messages.js.map