1234567891011module exchange1213import os14import src.events15import src.history16import src.remote17import src.snapshot18import src.storage192021const repo_state_key = 'remote_repo'22232425const remote_head_key = 'remote_head'2627pub const no_saves = 'this repo has no saves yet — nothing to offer'28293031const format_value = 'koh-remote 1'3233343536fn put_format(mut backend remote.Backend, repo string) ! {37 if backend.put_if_absent(format_key(repo), format_value.bytes())! {38 return39 }40 check_format(mut backend, repo)!41}4243444546fn check_format(mut backend remote.Backend, repo string) !string {47 existing := backend.get(format_key(repo))!48 if !existing.found {49 return ''50 }51 value := existing.body.bytestr().trim_space()52 if value != format_value {53 return error('the remote "${repo}" is "${value}" — this koh speaks "${format_value}"; update koh')54 }55 return value56}5758pub struct OfferReport {59pub mut:60 repo string61 saves int62 objects int63 events int64 head string65}66676869pub fn remember_repo(store &storage.Store, name string) ! {70 store.set_state(repo_state_key, name)!71}7273747576pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {77 if remembered := store.state(repo_state_key) {78 return remembered79 }80 name := os.base(store.root).to_lower()81 82 83 check_format(mut backend, name)!84 existing := backend.get(head_key(name))!85 if existing.found {86 theirs := existing.body.bytestr().trim_space()87 if history.by_id(store, theirs) == none {88 return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')89 }90 }91 store.set_state(repo_state_key, name)!92 return name93}94959697pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {98 name := repo_name(store, mut backend)!99 local := history.head(store) or { return error(no_saves) }100 mut report := OfferReport{101 repo: name102 head: local.id103 }104 if check_format(mut backend, name)! == '' {105 106 put_format(mut backend, name)!107 }108 remote_head := backend.get(head_key(name))!109 110 111 if unapplied_strips(store, mut backend, name)!.len > 0 {112 return error(strip_not_applied)113 }114 mut uploaded := Uploaded{}115 if remote_head.found {116 theirs := remote_head.body.bytestr().trim_space()117 newer := saves_after(store, local.id, theirs) or {118 return error('remote has new saves — run koh steal')119 }120 for save in newer {121 upload_save(store, mut backend, name, save, mut report, mut uploaded)!122 }123 } else {124 oldest_first := history.line(store)!125 for index := oldest_first.len - 1; index >= 0; index-- {126 upload_save(store, mut backend, name, oldest_first[index], mut report, mut uploaded)!127 }128 }129 upload_events(store, mut backend, name, mut report)!130 131 132 133 late := unapplied_strips(store, mut backend, name)!134 if late.len > 0 {135 take_back(store, mut backend, name, uploaded, late)!136 return error(strip_not_applied)137 }138 if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {139 return error('remote has new saves — run koh steal')140 }141 store.set_state(remote_head_key, local.id)!142 return report143}144145146147148fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {149 mut newer := []history.Save{}150 mut cursor := ours151 for cursor != '' && cursor != theirs {152 save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }153 newer << save154 cursor = save.parent155 }156 if cursor != theirs {157 return error('${theirs} is not on this line')158 }159 newer.reverse_in_place()160 return newer161}162163const strip_not_applied = 'the remote has a strip this repo has not applied — run koh steal'164165166struct Uploaded {167mut:168 saves []string169 objects map[string]bool170}171172173174fn take_back(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, strips []events.Event) ! {175 for id in uploaded.saves {176 for file in history.files_of(store, id)! {177 if file.hash in uploaded.objects && strips.any(reaches(it, id) && covers(it.path, file.path)) {178 backend.delete(object_key(repo, file.hash))!179 }180 }181 backend.delete(save_key(repo, id))!182 }183}184185186187188fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport, mut uploaded Uploaded) ! {189 files := history.files_of(store, save.id)!190 191 192 193 mut sent := map[string]bool{}194 if save.parent != '' {195 for file in history.files_of(store, save.parent)! {196 sent[file.hash] = true197 }198 }199 for file in files {200 if file.hash in sent {201 continue202 }203 sent[file.hash] = true204 if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {205 report.objects++206 uploaded.objects[file.hash] = true207 }208 }209 record := save_record(store, save)!210 if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {211 report.saves++212 uploaded.saves << save.id213 }214}215216217218fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {219 already := backend.list(events_prefix(repo))!220 mut already_sent := map[string]bool{}221 for key in already {222 already_sent[os.base(key)] = true223 }224 for event in events.all(store)! {225 if event.id in already_sent {226 continue227 }228 if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {229 report.events++230 }231 }232}233234235236237pub struct Fetched {238pub mut:239 repo_objects map[string][]u8240 saves []SaveRecord 241 events_out []events.Event242 head string243 base string 244 unsafe_paths []string245}246247248249250pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {251 check_format(mut backend, repo)!252 remote_head := backend.get(head_key(repo))!253 if !remote_head.found {254 return error('the remote has no "${repo}" yet — run koh offer here first')255 }256 mut fetched := Fetched{257 head: remote_head.body.bytestr().trim_space()258 }259 mut on_line := map[string]bool{}260 for save in history.line(store)! {261 on_line[save.id] = true262 }263 mut cursor := fetched.head264 for cursor != '' && cursor !in on_line {265 mut record := if saved := history.by_id(store, cursor) {266 save_record(store, saved)!267 } else {268 record_object := backend.get(save_key(repo, cursor))!269 if !record_object.found {270 return error('the remote is missing save ${cursor} — run koh offer there')271 }272 decode_save(record_object.body)!273 }274 fetched.unsafe_paths << drop_unsafe_paths(mut record)275 for file in record.files {276 if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects {277 blob := backend.get(object_key(repo, file.hash))!278 if !blob.found {279 return error('the remote is missing object ${file.hash} that save ${record.id} needs')280 }281 fetched.repo_objects[file.hash] = blob.body282 }283 }284 fetched.saves << record285 cursor = record.parent286 }287 fetched.saves.reverse_in_place()288 fetched.base = cursor289 fetched.events_out = fetch_events(store, mut backend, repo)!290 return fetched291}292293fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {294 have := events.all(store)!295 mut known := map[string]bool{}296 for event in have {297 known[event.id] = true298 }299 mut incoming := []events.Event{}300 for key in backend.list(events_prefix(repo))! {301 id := os.base(key)302 if id in known {303 continue304 }305 body := backend.get(event_key(repo, id))!306 if !body.found {307 continue308 }309 event := decode_event(body.body)!310 if !ignored_strip(event) {311 incoming << event312 }313 }314 return incoming315}316317318319320pub fn write_fetched(store &storage.Store, fetched Fetched) ! {321 for hash, blob in fetched.repo_objects {322 store.put_object_blob(hash, blob)!323 }324 mut stripped := false325 for event in fetched.events_out {326 if !events.receive(store, event)! {327 continue328 }329 if event.kind == events.kind_strip {330 remove_path(store, event.path, event.save_id)!331 stripped = true332 }333 }334 strips := known_strips(store, fetched)!335 for record in fetched.saves {336 import_arriving(store, record, strips)!337 }338 if stripped {339 store.drop_unreferenced_objects()!340 }341}342343344345pub fn write_fetched_records(store &storage.Store, fetched Fetched) ! {346 for hash, blob in fetched.repo_objects {347 store.put_object_blob(hash, blob)!348 }349 strips := known_strips(store, fetched)!350 for record in fetched.saves {351 import_arriving(store, record, strips)!352 }353}354355356357fn known_strips(store &storage.Store, fetched Fetched) ![]events.Event {358 mut strips := events.strip_events(store)!359 strips << fetched.events_out.filter(it.kind == events.kind_strip)360 return strips361}362363364365fn import_arriving(store &storage.Store, record SaveRecord, strips []events.Event) ! {366 if history.by_id(store, record.id) != none {367 return368 }369 files := to_files(record).filter(fn [strips, record] (file snapshot.FileEntry) bool {370 return snapshot.legal_path(file.path)371 && !strips.any(reaches(it, record.id) && covers(it.path, file.path))372 })373 history.import_save(store, to_save(record), files)!374}375376377378fn reaches(strip events.Event, save_id string) bool {379 return strip.save_id == '' || strip.save_id == save_id380}