//! Sync: offer and steal.
//!
//! All the cleverness sits here, above the five-operation remote interface, so
//! the backends stay thin and this logic is written once and tested offline.
//!
//! The remote is a line of saves, the same shape as the local one. `offer`
//! sends local saves the remote lacks; `steal` takes remote saves and, when
//! both sides moved, replays the local work on top of the remote's. Divergence
//! is resolved per file, never per line.

module exchange

import json2
import os
import src.events
import src.fmt
import src.history
import src.remote
import src.snapshot
import src.storage

// Where the repo's name on the remote is remembered.
const repo_state_key = 'remote_repo'

// The remote head as of this machine's last successful offer: a record for the
// person or a future feature. Nothing depends on it.
const remote_head_key = 'remote_head'

pub const no_saves = 'this repo has no saves yet — nothing to offer'

// The only remote layout this koh reads or writes. A repo offered before the
// key existed has no `format` key and counts as this version.
const format_value = 'koh-remote 1'

// put_format records the layout on a repo that has none. put_if_absent never
// overwrites, so a machine that wrote one while this one read wins the race:
// its value is read back and checked, never assumed.
fn put_format(mut backend remote.Backend, repo string) ! {
	if backend.put_if_absent(format_key(repo), format_value.bytes())! {
		return
	}
	check_format(mut backend, repo)!
}

// check_format reads <repo>/format and refuses a remote written by a koh that
// speaks another layout, so two versions never half-understand one bucket. An
// empty answer means the key is missing: version 1, which `offer` records.
fn check_format(mut backend remote.Backend, repo string) !string {
	existing := backend.get(format_key(repo))!
	if !existing.found {
		return ''
	}
	value := existing.body.bytestr().trim_space()
	if value != format_value {
		return error('the remote "${repo}" is "${value}" — this koh speaks "${format_value}"; update koh')
	}
	return value
}

pub struct OfferReport {
pub mut:
	repo    string
	saves   int
	objects int
	events  int
	head    string
}

// remember_repo binds a repo to its name on the remote. A steal into a new folder does it at
// once: its folder may be renamed before it ever offers.
pub fn remember_repo(store &storage.Store, name string) ! {
	store.set_state(repo_state_key, name)!
}

// repo_name is the folder's name at first offer, lowercased so it makes the same
// Kepr URL whatever the folder's case (June, 2026-10-01), and remembered in the
// repo so renaming the folder later changes nothing.
pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {
	if remembered := store.state(repo_state_key) {
		return follow_rename(store, mut backend, remembered)!
	}
	name := os.base(store.root).to_lower()
	// Before the clash check below: a remote this koh cannot read is a "update
	// koh" answer, not a "rename the folder" one.
	check_format(mut backend, name)!
	existing := backend.get(head_key(name))!
	if existing.found {
		theirs := existing.body.bytestr().trim_space()
		if history.by_id(store, theirs) == none {
			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')
		}
	}
	store.set_state(repo_state_key, name)!
	return name
}

// follow_rename follows the marker Kepr leaves when it renames a repo, `.kepr/renamed/<old>`
// holding the new name (§30, 2026-10-01), for as long as the remembered name has no `head`, and
// remembers where it led. A few hops cover a repo renamed more than once.
fn follow_rename(store &storage.Store, mut backend remote.Backend, remembered string) !string {
	mut name := remembered
	for _ in 0 .. 5 {
		if backend.get(head_key(name))!.found {
			break
		}
		marker := backend.get('.kepr/renamed/${name}')!
		next := marker.body.bytestr().trim_space()
		if !marker.found || next == '' || next.contains('/') || next.contains('\\') || next in [
			'.',
			'..',
		] {
			break
		}
		name = next
	}
	if name != remembered {
		store.set_state(repo_state_key, name)!
	}
	return name
}

// offer sends what the remote lacks and then swaps `head`. Nothing is ever
// queued: the next offer simply sends everything the remote does not have.
pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {
	name := repo_name(store, mut backend)!
	local := history.head(store) or { return error(no_saves) }
	mut report := OfferReport{
		repo: name
		head: local.id
	}
	if check_format(mut backend, name)! == '' {
		// Offered before this key existed: record the layout now.
		put_format(mut backend, name)!
	}
	remote_head := backend.get(head_key(name))!
	// Records offered by a machine that has not applied a strip would put the
	// stripped path back on the remote, where nothing would rewrite it again.
	if unapplied_strips(store, mut backend, name)!.len > 0 {
		return error(strip_not_applied)
	}
	mut uploaded := Uploaded{}
	upload_line(store, mut backend, name, local.id, remote_head, mut report, mut uploaded) or {
		return after_failed_upload(store, mut backend, name, uploaded, err)
	}
	// A strip sends its event before it lists the saves to rewrite, so one
	// missed here lists them after these records landed and rewrites them. One
	// seen here may have rewritten before they landed: take back what it reaches.
	if undo_late_strips(store, mut backend, name, uploaded)! {
		return error(strip_not_applied)
	}
	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {
		return error('remote has new saves — run koh steal')
	}
	store.set_state(remote_head_key, local.id)!
	return report
}

// saves_after lists the local saves newer than `theirs`, oldest first. It fails
// when `theirs` is not on this line, which is the honest answer: we cannot
// offer on top of a save we do not have.
fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {
	mut newer := []history.Save{}
	mut cursor := ours
	for cursor != '' && cursor != theirs {
		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }
		newer << save
		cursor = save.parent
	}
	if cursor != theirs {
		return error('${theirs} is not on this line')
	}
	newer.reverse_in_place()
	return newer
}

const strip_not_applied = 'the remote has a strip this repo has not applied — run koh steal'

// Uploaded is what one upload put on the remote, so a late strip can take it back.
struct Uploaded {
mut:
	saves   []string
	objects map[string]bool
	deleted int // damaged remote copies removed before their replacement
}

fn (u Uploaded) done() []string {
	mut done := done_text('deleted', u.deleted, 'damaged remote object')
	if u.saves.len + u.objects.len > 0 {
		done << 'uploaded ${fmt.plural(u.objects.len, 'object')} and ${fmt.plural(u.saves.len, 'save record')}'
	}
	return done
}

// upload_line sends every save after the remote head, then the events.
fn upload_line(store &storage.Store, mut backend remote.Backend, name string, local string, remote_head remote.Object, mut report OfferReport, mut uploaded Uploaded) ! {
	if remote_head.found {
		theirs := remote_head.body.bytestr().trim_space()
		newer := saves_after(store, local, theirs) or {
			return error('remote has new saves — run koh steal')
		}
		for save in newer {
			upload_save(store, mut backend, name, save, mut report, mut uploaded)!
		}
	} else {
		oldest_first := history.line(store)!
		for index := oldest_first.len - 1; index >= 0; index-- {
			upload_save(store, mut backend, name, oldest_first[index], mut report, mut uploaded)!
		}
	}
	upload_events(store, mut backend, name, mut report)!
}

// after_failed_upload still applies a strip that landed before the failure
// to what the upload had put.
fn after_failed_upload(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, failure IError) IError {
	undo_late_strips(store, mut backend, repo, uploaded) or {
		return error('${failure.msg()}; removing what a strip on the remote covers also failed: ${err.msg()}')
	}
	return failure
}

// unsettled_key keeps what an upload sent when settling a late strip failed,
// for Doctor to finish once a steal brings the strip here.
pub const unsettled_key = 'upload_unsettled'

pub fn unsettled(store &storage.Store) ![]string {
	text := store.state(unsettled_key) or { return []string{} }
	return json2.decode[[]string](text) or {
		return error('the note of an unsettled upload is unreadable; it is kept for recovery')
	}
}

fn undo_late_strips(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded) !bool {
	return settle_late_strips(store, mut backend, repo, uploaded) or {
		failure := err
		mut kept := unsettled(store) or { return error('${failure.msg()}; ${err.msg()}') }
		kept << uploaded.objects.keys().filter(it !in kept)
		store.set_state(unsettled_key, json2.encode(kept)) or {
			return error('${failure.msg()}; ${err.msg()}')
		}
		return failure
	}
}

// settle_late_strips applies strips that reached the remote during an upload
// to what it put, and reports whether there were any. Records stay, less the
// stripped paths, as another offer or head may name them. Notes go first and
// objects last, so a stop part way leaves a note and no record without its object.
fn settle_late_strips(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded) !bool {
	strips := unapplied_strips(store, mut backend, repo)!
	if strips.len == 0 {
		return false
	}
	for strip in strips {
		queue_strip(store, strip.path, [strip.save_id], uploaded.objects.keys())!
	}
	for id in uploaded.saves {
		strip_record(mut backend, save_key(repo, id), id, strips)!
	}
	drop_stripped_objects(store, mut backend, repo, uploaded, strips)!
	return true
}

// strip_record rewrites one remote record without what `strips` cover. A
// record that moved meanwhile is read again, since the other writer may have
// kept a path these strips name.
fn strip_record(mut backend remote.Backend, key string, id string, strips []events.Event) ! {
	for _ in 0 .. 5 {
		object := backend.get(key)!
		if !object.found {
			return
		}
		record := decode_save(object.body)!
		kept := record.files.filter(fn [strips, id] (file FileRecord) bool {
			return !strips.any(reaches(it, id) && covers(it.path, file.path))
		})
		if kept.len == record.files.len
			|| backend.put_if_match(key, encode_save(rewrite(record, kept)).bytes(), object.tag)! {
			return
		}
	}
	return error('the remote copy of save ${id} kept changing while a stripped path was removed; run koh doctor again')
}

// Partway is a failure after some work had already landed. `done` says what,
// so Doctor can report both.
pub struct Partway {
	Error
pub:
	done   string
	reason string
}

pub fn (e Partway) msg() string {
	return e.reason
}

// done_text is what a step finished before it failed, if anything.
pub fn done_text(verb string, done int, noun string) []string {
	return if done == 0 { []string{} } else { ['${verb} ${fmt.plural(done, noun)}'] }
}

// partway adds what was `done` before `failure` to it, including what a
// failing step of its own had done.
pub fn partway(done []string, failure IError) IError {
	mut all := done.clone()
	if failure is Partway {
		all << failure.done
	}
	if all.len == 0 {
		return failure
	}
	return Partway{
		done:   all.join('; ')
		reason: failure.msg()
	}
}

// drop_stripped_objects deletes what an upload put that a strip reaches in a
// save here, unless a remote record lists it under a path no strip covers:
// another machine may hold the same bytes there.
fn drop_stripped_objects(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, strips []events.Event) ! {
	mut reached := map[string]bool{}
	// A local save that still names it under another path may have no other copy.
	mut kept := map[string]bool{}
	for save in history.all_saves(store)! {
		for file in history.files_of(store, save.id)!.filter(it.hash in uploaded.objects) {
			if strips.any(reaches(it, save.id) && covers(it.path, file.path)) {
				reached[file.hash] = true
			} else {
				kept[file.hash] = true
			}
		}
	}
	if reached.len == 0 {
		return
	}
	for key in backend.list('${repo}/saves')! {
		object := backend.get(key)!
		if object.found {
			record := decode_save(object.body)!
			for file in record.files {
				if !strips.any(reaches(it, record.id) && covers(it.path, file.path)) {
					reached.delete(file.hash)
				}
			}
		}
	}
	for hash, _ in reached {
		if hash !in kept {
			backend.delete(object_key(repo, hash))!
		}
	}
}

// upload_save puts a save's new objects, then its record, so a record on the remote always finds its
// objects there. Both are put_if_absent, so a retry after a partial offer is safe; `uploaded` keeps
// what this call created, which is all undo_late_strips may delete.
fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport, mut uploaded Uploaded) ! {
	files := history.files_of(store, save.id)!
	// Saves go up oldest first and head moves only once all of them landed, so
	// the parent and its objects are on the remote already: only content new
	// in this save travels, not a round trip per file in the repo.
	mut sent := map[string]bool{}
	if save.parent != '' {
		for file in history.files_of(store, save.parent)! {
			sent[file.hash] = true
		}
	}
	for file in files {
		if file.hash in sent {
			continue
		}
		sent[file.hash] = true
		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {
			report.objects++
			uploaded.objects[file.hash] = true
		}
	}
	record := save_record(store, save)!
	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {
		report.saves++
		uploaded.saves << save.id
	}
}

// upload_events sends every event the remote does not have. Events are tiny, so
// listing them costs less than tracking which ones were sent.
fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {
	already := backend.list(events_prefix(repo))!
	mut already_sent := map[string]bool{}
	for key in already {
		already_sent[os.base(key)] = true
	}
	for event in events.all(store)! {
		if event.id in already_sent {
			continue
		}
		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {
			report.events++
		}
	}
}

// fetched holds everything a steal read from the remote, before any of it is
// written locally. Network work happens first; the database gets one
// transaction, so a failure leaves nothing half-applied.
pub struct Fetched {
pub mut:
	repo_objects map[string][]u8
	saves        []SaveRecord // oldest first
	events_out   []events.Event
	head         string
	base         string // the newest save both sides already shared, '' when none
	unsafe_paths []string
}

// fetch_back_to_known walks from the remote head along parents until it meets a save on this line
// (`base`), reading only the objects this repo lacks. Nothing is written here: write_fetched lands it.
// A save held locally but off the line, from an earlier steal, is taken from the local copy.
pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {
	check_format(mut backend, repo)!
	remote_head := backend.get(head_key(repo))!
	if !remote_head.found {
		return error('the remote has no "${repo}" yet — run koh offer here first')
	}
	mut fetched := Fetched{
		head: remote_head.body.bytestr().trim_space()
	}
	if !valid_save_id(fetched.head) {
		return error('the remote head has invalid save ID ${fetched.head}')
	}
	mut on_line := map[string]bool{}
	for save in history.line(store)! {
		on_line[save.id] = true
	}
	// Strips come first: an object only a stripped path named may be gone.
	fetched.events_out = fetch_events(store, mut backend, repo)!
	strips := known_strips(store, fetched)!
	mut cursor := fetched.head
	mut seen := map[string]bool{}
	for cursor != '' && cursor !in on_line {
		if cursor in seen {
			return error('the remote history has a parent cycle at ${cursor}')
		}
		seen[cursor] = true
		mut record := if saved := history.by_id(store, cursor) {
			save_record(store, saved)!
		} else {
			record_object := backend.get(save_key(repo, cursor))!
			if !record_object.found {
				return error('the remote is missing save ${cursor} — run koh offer there')
			}
			decoded := decode_save(record_object.body)!
			validate_save(decoded, cursor)!
			decoded
		}
		fetched.unsafe_paths << drop_unsafe_paths(mut record)
		validate_hashes(record.id, record.files)!
		for file in record.files {
			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects
				&& !strips.any(reaches(it, record.id) && covers(it.path, file.path)) {
				blob := backend.get(object_key(repo, file.hash))!
				if !blob.found {
					return error('the remote is missing object ${file.hash} that save ${record.id} needs')
				}
				fetched.repo_objects[file.hash] = blob.body
			}
		}
		fetched.saves << record
		cursor = record.parent
	}
	fetched.saves.reverse_in_place()
	fetched.base = cursor
	return fetched
}

fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {
	have := events.all(store)!
	mut known := map[string]bool{}
	for event in have {
		known[event.id] = true
	}
	mut incoming := []events.Event{}
	for key in backend.list(events_prefix(repo))! {
		id := os.base(key)
		if !valid_event_id(id) {
			return error('the remote has invalid event ID ${id}')
		}
		if id in known {
			continue
		}
		body := backend.get(event_key(repo, id))!
		if !body.found {
			continue
		}
		event := decode_event(body.body)!
		if event.id != id {
			return error('remote event record ID ${event.id} disagrees with key ${id}')
		}
		if !ignored_strip(event) {
			incoming << event
		}
	}
	return incoming
}

// write_fetched lands a fetch in one transaction: objects first (verified),
// then the events, then the saves they belong to. A strip bans its path from
// all of history, so it reaches every save, here or arriving, whenever made.
pub fn write_fetched(store &storage.Store, fetched Fetched) ! {
	for hash, blob in fetched.repo_objects {
		store.put_object_blob(hash, blob)!
	}
	mut freed := []string{}
	for event in fetched.events_out {
		if !events.receive(store, event)! {
			continue
		}
		if events.valid_strip(event) {
			freed << stripped_hashes(store, event.path, event.save_id)!
			remove_path(store, event.path, event.save_id)!
		}
	}
	strips := known_strips(store, fetched)!
	for record in fetched.saves {
		for file in record.files {
			if file.hash in fetched.repo_objects
				&& strips.any(reaches(it, record.id) && covers(it.path, file.path)) {
				freed << file.hash
			}
		}
		import_arriving(store, record, strips)!
	}
	store.drop_unused(freed)!
}

// write_fetched_records makes network work durable without changing the line.
// Events wait for the combine transaction because a strip can rewrite history.
pub fn write_fetched_records(store &storage.Store, fetched Fetched) ! {
	for hash, blob in fetched.repo_objects {
		store.put_object_blob(hash, blob)!
	}
	strips := known_strips(store, fetched)!
	for record in fetched.saves {
		import_arriving(store, record, strips)!
	}
}

// known_strips is every valid strip held here and every one the fetch brings:
// an arriving save loses what any of them reaches, not only what arrived with it.
fn known_strips(store &storage.Store, fetched Fetched) ![]events.Event {
	mut strips := events.strip_events(store)!
	strips << fetched.events_out
	return strips.filter(events.valid_strip(it))
}

// import_arriving lands one save from the remote, unless it is already here, less every path a
// strip that `reaches` it removed.
fn import_arriving(store &storage.Store, record SaveRecord, strips []events.Event) ! {
	if history.by_id(store, record.id) != none {
		return
	}
	validate_save(record, record.id)!
	safe := record.files.filter(snapshot.legal_path(it.path))
	validate_hashes(record.id, safe)!
	files := to_files(record).filter(fn [strips, record] (file snapshot.FileEntry) bool {
		return snapshot.legal_path(file.path)
			&& !strips.any(reaches(it, record.id) && covers(it.path, file.path))
	})
	history.import_save(store, to_save(record), files)!
}

// reaches reports whether a strip event applies to a save: its one save, or
// every save there is, whenever it was made.
fn reaches(strip events.Event, save_id string) bool {
	return strip.save_id == '' || strip.save_id == save_id
}