-- SPDX-FileCopyrightText: © 2026 Vladimir Zorin <vladimir@deviant.guru>
-- SPDX-License-Identifier: LicenseRef-OWL-1.0-or-later
-- Licensed under OWL v1.0+. See LICENSE.

local std = require("std")
local json = require("cjson.safe")
local schema = require("agent_smith.persistence.schema")
local repository = require("agent_smith.persistence.repository")

local function conversation_title(text)
	if type(text) ~= "string" then
		return nil
	end
	local trimmed = std.txt.trim(text:gsub("%s+", " "))
	if trimmed == "" then
		return nil
	end
	if std.utf.len(trimmed) > 80 then
		return std.utf.sub(trimmed, 1, 77) .. "..."
	end
	return trimmed
end

local function encode_json(value)
	local encoded = json.encode(value)
	if encoded then
		return encoded
	end
	return "{}"
end

local new = function(options)
	options = options or {}

	local repositories_handle = nil
	local turn_refs = {}

	local repositories = function()
		if repositories_handle then
			return repositories_handle
		end
		repositories_handle = repository.new(options.repository_options or {})
		return repositories_handle
	end

	local persist_turn = function(record)
		record = record or {}
		local turn_id = record.turn_id or ("turn-" .. tostring(#turn_refs + 1))
		local conversation_id = record.conversation_id or "conversation-default"
		local repos = repositories()

		local existing_conversation = repos.conversation.get_conversation(conversation_id)

		local existing, list_err = repos.conversation.list_messages(conversation_id)
		if not existing then
			return nil, list_err
		end
		-- Derive the next seq from the highest existing seq, not the message
		-- count: append_message is an upsert keyed by seq, so a sparse range
		-- (count < max(seq)+1) would otherwise overwrite a prior message.
		-- Rows are sorted by key, so the last row carries the highest seq.
		local next_seq = 0
		if #existing > 0 then
			local last_seq = existing[#existing].seq
			if type(last_seq) == "number" then
				next_seq = last_seq + 1
			else
				next_seq = #existing
			end
		end
		local message_index = 0
		local append_message = function(message)
			message_index = message_index + 1
			local message_id = turn_id .. "-msg-" .. tostring(message_index)
			local ok, err = repos.conversation.append_message({
				conversation_id = conversation_id,
				seq = next_seq,
				message_id = message_id,
				role = message.role,
				content = message.content,
				tool_call_id = message.tool_call_id,
				tool_calls = message.tool_calls,
			})
			if not ok then
				return nil, err
			end
			next_seq = next_seq + 1
			return true
		end

		-- Write the turn's transcript slice verbatim (user message, then each
		-- round's assistant tool_calls / tool results / assistant text in true
		-- order), so the persisted conversation matches the live one.
		for i = 1, #(record.messages or {}) do
			local message = record.messages[i]
			if type(message) == "table" and type(message.role) == "string" then
				local ok, err = append_message({
					role = message.role,
					content = message.content,
					tool_call_id = message.tool_call_id,
					tool_calls = message.tool_calls,
				})
				if not ok then
					return nil, err
				end
			end
		end

		-- Persist/merge conversation metadata now that messages are appended and
		-- the count is known — BEFORE the usage/note writes. The conversation
		-- record + transcript are the durable state /history and /load depend on,
		-- so writing them ahead of the more fragile usage/note steps guarantees a
		-- turn is visible in /history even if a later step errors. Preserve
		-- created_at/title/project_id across turns; refresh updated_at/message_count.
		-- Wall-clock epoch seconds for display/sorting in /history. (lev.now() is a
		-- monotonic uptime clock, not epoch — using it yielded 1970-era dates.)
		local now = os.time()
		local conversation_record = {}
		if type(existing_conversation) == "table" then
			for k, v in pairs(existing_conversation) do
				conversation_record[k] = v
			end
		end
		conversation_record.conversation_id = conversation_id
		conversation_record.updated_at = now
		conversation_record.message_count = next_seq
		if conversation_record.created_at == nil then
			conversation_record.created_at = now
		end
		if
			conversation_record.project_id == nil
			and type(record.project_id) == "string"
			and record.project_id ~= ""
		then
			conversation_record.project_id = record.project_id
		end
		if conversation_record.title == nil then
			local title = conversation_title(record.user_message)
			if title then
				conversation_record.title = title
			end
		end
		local _, conv_err = repos.conversation.update_conversation(conversation_record)
		if conv_err then
			return nil, conv_err
		end

		-- Remember this as the active conversation so the next launch can
		-- auto-resume it. Best-effort: a failure here must not fail the turn,
		-- since the conversation + transcript are already durably written.
		local active_key = schema.build_meta_active_conversation_id_key()
		repos.adapter:put(schema.KEYSPACE.meta, active_key, conversation_id)

		local usage = record.usage_delta or {}
		local usage_ok, usage_err = repos.usage.record_parent({
			conversation_id = conversation_id,
			turn_id = turn_id,
			prompt_tokens = usage.input_tokens,
			completion_tokens = usage.output_tokens,
			total_tokens = usage.total_tokens,
		})
		if not usage_ok then
			return nil, usage_err
		end

		-- Record child dispatch usage
		if record.tool_calls_executed then
			for _, tc in ipairs(record.tool_calls_executed) do
				if type(tc) == "table" and type(tc.content) == "table" and type(tc.content.usage) == "table" then
					local child_usage = tc.content.usage
					-- Key child usage by the real dispatch job id (stamped onto the
					-- dispatch_result by the dispatch tool). build_usage_child_key
					-- requires a colon-free segment, so fall back to a fresh nanoid
					-- only when a usable real id is missing.
					local job_id = tc.content.job_id
					if type(job_id) ~= "string" or job_id == "" or string.find(job_id, ":", 1, true) then
						job_id = std.nanoid()
					end
					local child_ok, child_err = repos.usage.record_child({
						conversation_id = conversation_id,
						turn_id = turn_id,
						job_id = job_id,
						prompt_tokens = child_usage.input_tokens,
						completion_tokens = child_usage.output_tokens,
						total_tokens = child_usage.total_tokens,
					})
					if not child_ok then
						return nil, child_err
					end
				end
			end
		end

		local note_key = "turn_record-" .. turn_id
		local encoded = json.encode(record)
		if not encoded then
			-- Full turn record failed to encode; persist a minimal stub flagged so
			-- the data loss is visible rather than silently discarded.
			encoded = encode_json({
				turn_id = turn_id,
				conversation_id = conversation_id,
				stop_reason = record.stop_reason,
				cancelled = record.cancelled,
				encoding_error = "full turn record failed to JSON-encode",
			})
		end
		local _, note_err = repos.conversation_note.upsert({
			conversation_id = conversation_id,
			note_id = note_key,
			content = encoded,
		})
		if note_err then
			return nil, note_err
		end

		turn_refs[#turn_refs + 1] = {
			conversation_id = conversation_id,
			turn_id = turn_id,
		}
		return { ok = true, turn_id = turn_id }
	end

	local purge_conversation = function(conversation_id)
		if type(conversation_id) ~= "string" or conversation_id == "" then
			return nil, "conversation_id must be a non-empty string"
		end
		local repos = repositories()
		local adapter = repos.adapter
		if type(adapter) ~= "table" then
			return nil, "persistence adapter unavailable"
		end

		-- Conversation record lives at an exact key; prefix-scanning could match
		-- sibling ids that share a prefix, so target it directly.
		local conv_key, conv_key_err = schema.build_conversation_key(conversation_id)
		if not conv_key then
			return nil, conv_key_err
		end

		-- Collect every delete into a single batch so the purge is atomic: a
		-- failure partway through rolls back rather than leaving a half-purged
		-- conversation.
		local ops = {
			{ op = "delete", keyspace = schema.KEYSPACE.conversation, key = conv_key },
		}

		-- Every other conversation-scoped record shares the "<id>:..." prefix.
		local prefix, prefix_err = schema.build_conversation_scope_prefix(conversation_id)
		if not prefix then
			return nil, prefix_err
		end
		local scoped = {
			schema.KEYSPACE.message,
			schema.KEYSPACE.message_search,
			schema.KEYSPACE.conversation_note,
			schema.KEYSPACE.usage,
		}
		for _, keyspace in ipairs(scoped) do
			local rows, list_err = adapter:list(keyspace, prefix)
			if not rows then
				return nil, list_err
			end
			for i = 1, #rows do
				ops[#ops + 1] = { op = "delete", keyspace = keyspace, key = rows[i].key }
			end
		end

		-- If this is the active conversation, drop the auto-resume pointer too so
		-- the next launch does not try to resume a conversation we just deleted.
		local active_key = schema.build_meta_active_conversation_id_key()
		local active_value, active_err = adapter:get(schema.KEYSPACE.meta, active_key)
		if active_err and active_err ~= "not found" then
			return nil, active_err
		end
		if active_value == conversation_id then
			ops[#ops + 1] = { op = "delete", keyspace = schema.KEYSPACE.meta, key = active_key }
		end

		local _, batch_err = adapter:batch(ops)
		if batch_err then
			return nil, batch_err
		end

		-- Drop in-memory turn refs for this conversation so turn_records() does
		-- not try to resurrect purged notes.
		local kept = {}
		for i = 1, #turn_refs do
			if turn_refs[i].conversation_id ~= conversation_id then
				kept[#kept + 1] = turn_refs[i]
			end
		end
		turn_refs = kept

		return { ok = true }
	end

	local turn_records = function()
		local repos = repositories()
		local out = {}
		for i = 1, #turn_refs do
			local ref = turn_refs[i]
			local note = repos.conversation_note.get(ref.conversation_id, "turn_record-" .. ref.turn_id)
			if note then
				local decoded = json.decode(note.content)
				if decoded then
					out[#out + 1] = decoded
				end
			end
		end
		return out
	end

	-- Conversation id of the most recently persisted turn, or nil if none.
	-- Stored as a raw meta value (like schema_version), not a codec record.
	local get_active_conversation_id = function()
		local repos = repositories()
		local key = schema.build_meta_active_conversation_id_key()
		local value, err = repos.adapter:get(schema.KEYSPACE.meta, key)
		if err == "not found" then
			return nil
		end
		if err then
			return nil, err
		end
		if type(value) ~= "string" or value == "" then
			return nil
		end
		return value
	end

	local close = function()
		if repositories_handle and repositories_handle.close then
			return repositories_handle.close()
		end
		return true
	end

	return {
		name = "persistence",
		kind = "service",
		options = options,
		persist_turn = persist_turn,
		purge_conversation = purge_conversation,
		turn_records = turn_records,
		get_active_conversation_id = get_active_conversation_id,
		repositories = repositories,
		close = close,
	}
end

return { new = new }