//! Stealing: bringing another machine's saves in, and combining when both
//! machines moved.
//!
//! The order is always the same, and it is what makes an interrupted steal
//! harmless: read everything from the remote first, then land it in one
//! transaction, then write the working tree. Divergence is resolved per file —
//! a clash is never merged line by line, it is chosen.

module sync

import compress.zlib
import os
import src.apply
import src.events
import src.history
import src.remote
import src.snapshot
import src.storage

pub enum Side {
	mine
	theirs
}

// Clash is one file both machines changed differently. `stat` is the short
// diff stat the terminal prompt shows, so the choice is informed.
pub struct Clash {
pub mut:
	path   string
	mine   snapshot.FileEntry
	theirs snapshot.FileEntry
	base   snapshot.FileEntry
	stat   string
}

pub struct StealOptions {
pub mut:
	mine   bool // resolve every clash as mine
	theirs bool // resolve every clash as theirs
	// Asked once per clash when neither flag is set. Nil means "nobody to ask".
	ask fn (clash Clash) Side = unsafe { nil }
}

pub struct StealReport {
pub mut:
	repo            string
	forward         bool
	fetched_saves   int
	fetched_objects int
	fetched_events  int
	clashes         []string
	kept_mine       int
	replayed        int
	removed         int
	head            string
}

pub const dirty_tree = 'save or checkpoint first — steal needs a clean tree'
pub const no_saves_here = 'this folder has files but no saves — steal a repo into an empty folder with "koh steal <name>"'

// steal brings the remote's new saves in. Inside a repo with its own work this
// may have to combine; a repo with nothing of its own simply fast-forwards.
pub fn steal(store &storage.Store, mut backend remote.Backend, opts StealOptions) !StealReport {
	name := repo_name(store, mut backend)!
	head_before := history.head(store)
	// Combining rewrites the working tree, so unsaved work would be lost.
	if dirty(store, head_before)! {
		if head_before == none {
			return error(no_saves_here)
		}
		return error(dirty_tree)
	}
	fetched := fetch_back_to_known(store, mut backend, name)!
	mut report := StealReport{
		repo:            name
		fetched_saves:   fetched.saves.len
		fetched_objects: fetched.repo_objects.len
		fetched_events:  fetched.events_out.len
		head:            fetched.head
	}
	mut local_only := []history.Save{}
	if head := head_before {
		// The remote builds on `base`. If that save is here but no longer on
		// the line, neither line contains the other; jumping to the remote
		// would silently leave this line behind.
		local_only = saves_after(store, head.id, fetched.base) or {
			return error('the remote builds on save ${fetched.base}, which is not on this line — load the save you want, then offer')
		}
	}
	if fetched.base == '' && local_only.len > 0 {
		return error('the remote keeps a different history — it shares no save with this repo')
	}

	strips_before := events.strip_events(store)!.len
	mut clashes := []Clash{}
	mut side_of := map[string]Side{}
	if fetched.saves.len > 0 && local_only.len > 0 {
		clashes = combine_clashes(store, fetched)!
		report.clashes = clashes.map(it.path)
		// Asked before the write lock: a person thinking must not stall every
		// agent's save on this repo.
		side_of = choose_sides(clashes, opts)!
	}

	store.begin() or { return err }
	// The questions ran without the lock. If a save or an edit landed
	// meanwhile, the answers were about a line that no longer exists.
	if moved := changed_since(store, head_before, strips_before) {
		if moved {
			store.rollback()
			return error('this repo changed while steal was asking — run koh steal again')
		}
	} else {
		store.rollback()
		return err
	}
	final := (write_and_combine(store, fetched, local_only, clashes, side_of, mut report) or {
		store.rollback()
		return err
	})
	store.commit() or { return err }
	report.head = final.id

	if head := head_before {
		if final.id == head.id {
			// The line did not move, so neither does the working tree.
			return report
		}
		files := history.files_of(store, final.id)!
		mut wanted := map[string]bool{}
		for file in files {
			wanted[file.path] = true
		}
		mut remove := []string{}
		for file in history.files_of(store, head.id)! {
			if file.path !in wanted {
				remove << file.path
			}
		}
		result := snapshot.restore_files(store, store.root, snapshot.load(store.root), files,
			remove)!
		report.removed = result.removed
	} else {
		files := history.files_of(store, final.id)!
		snapshot.restore_files(store, store.root, snapshot.load(store.root), files, [])!
	}
	history.write_log_md(store)!
	return report
}

// write_and_combine lands a fetch and, when the local line has saves the remote
// lacks, replays them on top of the remote head. It runs inside the caller's
// transaction and returns the new head.
fn write_and_combine(store &storage.Store, fetched Fetched, local_only []history.Save, clashes []Clash, side_of map[string]Side, mut report StealReport) !history.Save {
	write_fetched(store, fetched)!
	if fetched.saves.len == 0 {
		// The remote head is already on this line: level, or this machine is
		// ahead and the offer that follows a steal sends the rest.
		return history.head(store) or { return error('this repo has no head') }
	}
	if local_only.len == 0 {
		// Nothing of ours is missing from the remote: the line simply moves on.
		history.set_head(store, fetched.head)!
		report.forward = true
		return history.head(store) or { return error('the remote head did not land') }
	}
	remote_files := history.files_of(store, fetched.head)!
	mut mine_files := []snapshot.FileEntry{}
	if our_head := history.head(store) {
		mine_files = history.files_of(store, our_head.id)!
	}

	// Take the remote's line first, so the replayed saves chain onto it.
	history.set_head(store, fetched.head)!
	mut current := remote_files.clone()
	for save in local_only {
		parent_files := if save.parent != '' {
			history.files_of(store, save.parent)!
		} else {
			[]snapshot.FileEntry{}
		}
		target_files := history.files_of(store, save.id)!
		plan := apply.three_way(current, parent_files, target_files, [], clashes.map(it.path))
		if apply.conflicts(plan) {
			return error('combine could not apply ${plan.conflict} — it is not a clash file')
		}
		if plan.applied.len == 0 {
			// Everything this save touched was a clash resolved as theirs.
			continue
		}
		history.record_save(store, '', plan.files, history.SaveInput{
			message:       save.message
			created_at:    save.created_at
			machine:       save.machine
			agent:         save.agent
			session:       save.session
			is_checkpoint: save.is_checkpoint
			replayed_from: save.id
		})!
		current = plan.files.clone()
		report.replayed++
	}

	mut mine_paths := []string{}
	for clash in clashes {
		if side_of[clash.path] == .mine {
			mine_paths << clash.path
		}
	}
	if mine_paths.len > 0 {
		// One last save puts my versions of the chosen files back.
		final_plan := apply.three_way(current, current, mine_files, mine_paths, [])
		if final_plan.applied.len > 0 {
			history.record_save(store, '', final_plan.files, history.SaveInput{
				message: 'combine: kept mine for ${mine_paths.len} ${if mine_paths.len == 1 {
					'file'
				} else {
					'files'
				}}'
			})!
			report.kept_mine = mine_paths.len
		}
	}
	return history.head(store) or { return error('the combined line has no head') }
}

// find_clashes lists the files both sides changed, differently. Files changed
// identically on both sides are not clashes: there is nothing to decide.
pub fn find_clashes(base []snapshot.FileEntry, mine []snapshot.FileEntry, theirs []snapshot.FileEntry) []Clash {
	base_map := snapshot.index_files(base)
	mine_map := snapshot.index_files(mine)
	theirs_map := snapshot.index_files(theirs)
	mut paths := mine_map.keys()
	for path in theirs_map.keys() {
		if path !in mine_map {
			paths << path
		}
	}
	paths.sort()
	mut clashes := []Clash{}
	for path in paths {
		here := mine_map[path] or {
			snapshot.FileEntry{
				path: path
			}
		}
		there := theirs_map[path] or {
			snapshot.FileEntry{
				path: path
			}
		}
		was := base_map[path] or {
			snapshot.FileEntry{
				path: path
			}
		}
		if same(here, there) {
			continue
		}
		if same(here, was) || same(there, was) {
			continue
		}
		clashes << Clash{
			path:   path
			mine:   here
			theirs: there
			base:   was
		}
	}
	return clashes
}

// combine_clashes finds the files both lines changed, with the diff stat each
// prompt shows. It reads the remote head from the fetch, not the store, so it
// can run before anything is written.
fn combine_clashes(store &storage.Store, fetched Fetched) ![]Clash {
	base_files := if fetched.base != '' {
		history.files_of(store, fetched.base)!
	} else {
		[]snapshot.FileEntry{}
	}
	head := history.head(store) or { return []Clash{} }
	// The transaction applies the incoming strips before it replays, so the
	// clashes are found on the trees as they will be then, or the answers
	// would be about files that are no longer there.
	strips := fetched.events_out.filter(it.kind == events.kind_strip)
	mut clashes := find_clashes(after_strips(base_files, fetched.base, strips), after_strips(history.files_of(store,
		head.id)!, head.id, strips), after_strips(to_files(fetched.saves.last()), fetched.head,
		strips))
	for mut clash in clashes {
		mine := fetched_bytes(store, fetched, clash.mine)!
		theirs := fetched_bytes(store, fetched, clash.theirs)!
		delta := snapshot.diff_bytes(clash.path, 'modified', mine, theirs)!
		clash.stat = if delta.binary { 'binary' } else { '+${delta.added} -${delta.removed}' }
	}
	return clashes
}

// after_strips is a save's file list once the given strip events are applied.
fn after_strips(files []snapshot.FileEntry, save_id string, strips []events.Event) []snapshot.FileEntry {
	return files.filter(fn [strips, save_id] (file snapshot.FileEntry) bool {
		for strip in strips {
			if (strip.save_id == '' || strip.save_id == save_id) && covers(strip.path, file.path) {
				return false
			}
		}
		return true
	})
}

// fetched_bytes reads one side of a clash: from the fetch when it has just
// arrived, else from the store. An absent side is empty.
fn fetched_bytes(store &storage.Store, fetched Fetched, entry snapshot.FileEntry) ![]u8 {
	if entry.hash == '' {
		return []u8{}
	}
	if blob := fetched.repo_objects[entry.hash] {
		return zlib.decompress(blob)!
	}
	return store.get_object(entry.hash)!
}

fn choose_sides(clashes []Clash, opts StealOptions) !map[string]Side {
	mut sides := map[string]Side{}
	if clashes.len == 0 {
		return sides
	}
	for clash in clashes {
		if opts.mine {
			sides[clash.path] = .mine
			continue
		}
		if opts.theirs {
			sides[clash.path] = .theirs
			continue
		}
		if opts.ask != unsafe { nil } {
			sides[clash.path] = opts.ask(clash)
			continue
		}
		return error('${clashes.len} files changed on both machines — run on a terminal to choose, or pass --mine or --theirs')
	}
	return sides
}

fn same(a snapshot.FileEntry, b snapshot.FileEntry) bool {
	if a.hash == '' && b.hash == '' {
		return true
	}
	return a.hash == b.hash && a.executable == b.executable
}

fn changed_since(store &storage.Store, before ?history.Save, strips_before int) !bool {
	if events.strip_events(store)!.len != strips_before {
		return true
	}
	now := history.head(store)
	if now or { history.Save{} }.id != before or { history.Save{} }.id {
		return true
	}
	return dirty(store, now)!
}

// dirty reports whether the working tree holds anything a steal would
// overwrite. With no saves yet, that is every file but the untouched ignore
// list `koh init` wrote.
fn dirty(store &storage.Store, head ?history.Save) !bool {
	files := snapshot.walk(store, store.root, snapshot.load(store.root))!
	if saved := head {
		return !snapshot.same_tree(history.files_of(store, saved.id)!, files)
	}
	seed := storage.hash_hex(snapshot.default_ignore.bytes())
	return files.any(it.path != snapshot.ignore_file || it.hash != seed)
}

// fetch_repo pulls a whole repo — head, saves, objects and events — for
// `koh steal <name>`, which lands a copy in a new folder.
pub fn fetch_repo(mut backend remote.Backend, name string) !Fetched {
	remote_head := backend.get(head_key(name))!
	if !remote_head.found {
		return error('the remote has no repo called "${name}"')
	}
	save_ids := backend.list('${name}/saves')!
	object_keys := backend.list('${name}/objects')!
	event_keys := backend.list('${name}/events')!
	mut fetched := Fetched{
		head:            remote_head.body.bytestr().trim_space()
		remote_save_ids: map[string]bool{}
	}
	for key in object_keys {
		blob := backend.get(key)!
		if blob.found {
			fetched.repo_objects[os.base(key)] = blob.body
		}
	}
	for key in save_ids {
		body := backend.get(key)!
		if !body.found {
			continue
		}
		record := decode_save(body.body)!
		fetched.remote_save_ids[record.id] = true
		fetched.saves << record
	}
	fetched.saves = parent_first(fetched.saves)!
	for key in event_keys {
		body := backend.get(key)!
		if body.found {
			fetched.events_out << decode_event(body.body)!
		}
	}
	return fetched
}

// parent_first orders saves so every parent lands before its children, as the
// foreign key on `parent` demands. Time cannot do it: a replayed save keeps its
// original, older time but sits on a newer parent.
fn parent_first(records []SaveRecord) ![]SaveRecord {
	mut by_id := map[string]SaveRecord{}
	for record in records {
		by_id[record.id] = record
	}
	mut placed := map[string]bool{}
	mut ordered := []SaveRecord{}
	for record in records {
		mut chain := []SaveRecord{}
		mut cursor := record.id
		for cursor != '' && cursor !in placed {
			if chain.len > records.len {
				return error('the save records on the remote form a loop')
			}
			current := by_id[cursor] or { return error('the remote is missing save ${cursor}') }
			chain << current
			cursor = current.parent
		}
		chain.reverse_in_place()
		for current in chain {
			placed[current.id] = true
			ordered << current
		}
	}
	return ordered
}