//! Strip: removing a path from history, here and on the remote.
//!
//! The local half is one transaction: the saved files go, the objects nobody
//! references go, a strip event is recorded, and so is a note of which objects
//! the remote may now drop. The remote half reads that note, so it can be
//! re-run after an interruption even though the local rows are long gone.

module exchange

import json2
import os
import src.events
import src.remote
import src.storage

// Scope is what a strip would remove, for the warning shown before it runs.
pub struct Scope {
pub:
	files   int
	saves   int
	objects int
}

pub struct LocalStrip {
pub:
	files_removed   int
	objects_dropped i64
}

// SendReport says what a strip managed to make visible on the remote.
pub struct SendReport {
pub mut:
	records    int
	objects    int
	event_sent bool
}

// Pending is the note a strip leaves until its remote half has finished: which
// saves it covered ('' for every save) and which objects it freed. One note per
// path, so strips of the same path that did not finish are finished together.
struct Pending {
mut:
	saves  []string
	hashes []string
}

fn pending_key(path string) string {
	return 'strip_pending:${path}'
}

fn read_pending(store &storage.Store, path string) Pending {
	text := store.state(pending_key(path)) or { return Pending{} }
	return json2.decode[Pending](text) or { Pending{} }
}

// strip_pending is true while the remote half of a strip has not finished.
pub fn strip_pending(store &storage.Store, path string) bool {
	return store.state(pending_key(path)) != none
}

// covers is the one rule for what a strip removes: the path itself, or
// anything beneath it when it names a folder.
fn covers(path string, file string) bool {
	return file == path || file.starts_with('${path.trim_right('/')}/')
}

// strip_scope counts the rows strip_rows will delete, using the same match. `objects` is the distinct
// content they hold; content another path still uses is kept, so fewer may be deleted.
pub fn strip_scope(store &storage.Store, path string, only_save string) !Scope {
	mut stmt := store.db.prepare('SELECT COUNT(*), COUNT(DISTINCT save_id), COUNT(DISTINCT hash)
		FROM save_files WHERE ${match_clause(only_save)}')!
	defer {
		stmt.finalize()
	}
	bind_match(stmt, path, only_save)!
	stmt.step()!
	return Scope{
		files:   int(stmt.int(0))
		saves:   int(stmt.int(1))
		objects: int(stmt.int(2))
	}
}

// strip_local removes the path in one transaction and leaves the note the
// remote half needs.
pub fn strip_local(store &storage.Store, path string, only_save string) !LocalStrip {
	store.begin()!
	report := strip_rows(store, path, only_save) or {
		store.rollback()
		return err
	}
	store.end()!
	return report
}

/// Strip the batch in one transaction; failures leave history unchanged.
pub fn strip_local_batch(store &storage.Store, paths []string, only_save string) ![]LocalStrip {
	store.begin()!
	mut reports := []LocalStrip{}
	for path in paths {
		reports << strip_rows(store, path, only_save) or {
			store.rollback()
			return err
		}
	}
	store.end() or {
		store.rollback()
		return err
	}
	return reports
}

// strip_rows is the body of strip_local's transaction. Its event has no save id for a whole-history
// strip (spec §10.2), and that empty id is what bans the path from every later walk.
fn strip_rows(store &storage.Store, path string, only_save string) !LocalStrip {
	mut hashes := []string{}
	mut stmt :=
		store.db.prepare('SELECT DISTINCT hash FROM save_files WHERE ${match_clause(only_save)}')!
	bind_match(stmt, path, only_save)!
	for stmt.step()! {
		hashes << stmt.text(0)
	}
	stmt.finalize()
	if hashes.len == 0 {
		// Already stripped: keep the pending note an unfinished remote needs.
		return LocalStrip{}
	}
	removed := remove_path(store, path, only_save)!
	dropped := store.drop_unreferenced_objects()!
	events.strip(store, only_save, path)!
	// An earlier strip of the same path may not have finished on the remote,
	// so the note grows rather than resets.
	mut note := read_pending(store, path)
	if only_save !in note.saves {
		note.saves << only_save
	}
	note.hashes << hashes
	store.set_state(pending_key(path), json2.encode(note))!
	return LocalStrip{
		files_removed:   removed
		objects_dropped: dropped
	}
}

// send_strip makes a strip visible on the remote: the event travels like any
// other, every remote save record listing the path is rewritten, and objects
// the strip freed are deleted once no remote record and no local save needs
// them. Every step is safe to repeat, and the pending note is cleared only
// when all of them have succeeded.
pub fn send_strip(store &storage.Store, mut backend remote.Backend, path string) !SendReport {
	name := repo_name(store, mut backend)!
	mut report := SendReport{}
	mut upload := OfferReport{
		repo: name
	}
	upload_events(store, mut backend, name, mut upload)!
	report.event_sent = upload.events > 0
	note := read_pending(store, path)
	mut freed := map[string]bool{}
	for hash in note.hashes {
		freed[hash] = true
	}
	every_save := '' in note.saves
	mut needed := map[string]bool{}
	for key in backend.list('${name}/saves')! {
		remote_object := backend.get(key)!
		if !remote_object.found {
			continue
		}
		record := decode_save(remote_object.body)!
		mut kept := []FileRecord{}
		for file in record.files {
			if (every_save || record.id in note.saves) && covers(path, file.path) {
				freed[file.hash] = true
				continue
			}
			kept << file
			needed[file.hash] = true
		}
		if kept.len == record.files.len {
			continue
		}
		if !backend.put_if_match(key, encode_save(rewrite(record, kept)).bytes(), remote_object.tag)! {
			return error('the remote copy of save ${record.id} moved while stripping — run koh steal, then strip again')
		}
		report.records++
	}
	for hash, _ in freed {
		if hash in needed || store.has_object(hash)! {
			continue
		}
		backend.delete(object_key(name, hash))!
		report.objects++
	}
	store.remove_state(pending_key(path))!
	return report
}

// unapplied_strips are the strip events on the remote this repo has not
// received yet. offer asks before uploading and again after, because a strip that lands
// in between would otherwise leave the stripped path in the bucket.
fn unapplied_strips(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {
	mut known := map[string]bool{}
	for event in events.all(store)! {
		known[event.id] = true
	}
	mut strips := []events.Event{}
	for key in backend.list(events_prefix(repo))! {
		if os.base(key) in known {
			continue
		}
		body := backend.get(key)!
		if !body.found {
			continue
		}
		event := decode_event(body.body)!
		if event.kind == events.kind_strip && !ignored_strip(event) {
			strips << event
		}
	}
	return strips
}

// rewrite returns the save record with a new file list.
fn rewrite(record SaveRecord, files []FileRecord) SaveRecord {
	return SaveRecord{
		...record
		files: files
	}
}

// remove_path deletes the rows and counts them; the objects go afterwards, once nothing references them.
fn remove_path(store &storage.Store, path string, only_save string) !int {
	mut stmt := store.db.prepare('DELETE FROM save_files WHERE ${match_clause(only_save)}')!
	defer {
		stmt.finalize()
	}
	bind_match(stmt, path, only_save)!
	stmt.step()!
	return store.db.changes()
}

// match_clause is `covers` in SQL. LIKE would ignore ASCII case and treat `_`
// and `%` as wildcards; a strip destroys, so the prefix is compared exactly.
fn match_clause(only_save string) string {
	scope := if only_save == '' { '' } else { ' AND save_id = ?3' }
	return '(path = ?1 OR substr(path, 1, length(?2)) = ?2)${scope}'
}

// bind_match fills what match_clause names: ?1 the path, ?2 its folder prefix, ?3 the save.
fn bind_match(stmt &storage.Stmt, path string, only_save string) ! {
	stmt.bind_text(1, path)!
	stmt.bind_text(2, '${path.trim_right('/')}/')!
	if only_save != '' {
		stmt.bind_text(3, only_save)!
	}
}