ConcurrentDownloader.par

type Event = either {
  .started(Nat) Os.Path,    // ("total size") "destination"
  .chunk(Nat) Os.Path,      // ("chunk size") "destination"
  .finished Os.Path,        // "path"
  .failed(String) Os.Path,  // ("error message") "destination"
}

def Main =
  let rootPath = Os.Path(".").append("Downloads") in
  rootPath                                             // : Os.Path
    ->ServeUploadsAndYieldDownloads("127.0.0.1:3000")  // : List<(Url) Os.Path>
    ->Map1(box DownloadFile)                           // : List<List<Event>>
    ->FanIn1                                           // : List<Event>
    ->ShowDownloadProgress(rootPath)                   // : !

def Map1 = Map(type (Url) Os.Path, List<Event>)
def FanIn1 = FanIn(type Event)

dec ServeUploadsAndYieldDownloads : [Os.Path, String] List<(Url) Os.Path>
def ServeUploadsAndYieldDownloads = [rootPath, listenUrl] chan yield {
  Debug.Log(Concat(*("Listening on ", listenUrl, "...")))
  catch e => {
    Debug.Log(Concat(*("Listening error: ", e)))
    yield.end!
  }

  Http.Listen(listenUrl).begin.case {
    .shutdown try! => { yield.end! }

    .incoming(request, respond) next => {
      let (method, url, headers) body = request
      catch e => {
        respond((400, *()) Bytes.Reader(e))
        next.loop
      }

      let path = rootPath.append(url.path->TrimStart("/"))

      String.Equals(method, "POST").case {
        .true! => {
          let try downloadUrl = catch e => .err e in
            String.ParserFromReader(type Http.Error)(body)
              .remainder.try
              ->Url.FromString

          yield.item((downloadUrl) path)
          respond((200, *()) Bytes.EmptyReader)
          next.loop
        }
        .false! => {}
      }

      String.Equals(method, "GET").case {
        .true! => {
          body.close
          let try sourceFile = path.openFile
          respond((200, *()) sourceFile)
          next.loop
        }
        .false! => {}
      }

      body.close
      throw "Wrong method"
    }
  }
}

dec DownloadFile : [(Url) Os.Path] List<Event>
def DownloadFile = [(url) dst] chan yield {
  catch err => {
    yield.item(.failed(err) dst)
    yield.end!
  }

  let try (status, headers) body = Http.Fetch(("GET", url, *()) Bytes.EmptyReader)
  let headers = BoxMap.String(type Bytes)(headers)
  catch/clearResp err => { body.close; throw err }

  Nat.Equals(status, 200).case {
    .false! => { throw/clearResp "Bad server response" }
    .true! => {}
  }

  let try/clearResp file = dst.createNewFile
  catch/clearFile err => { file.close; throw err }

  let default(0) totalSize = catch ! => .err! in
    headers.get("Content-Length").try
      ->String.FromBytes
      ->Nat.FromString
  yield.item(.started(totalSize) dst)

  let accum = 0

  body.begin.read.try/clearFile.case {
    .chunk(bytes) => {
      file.write(bytes).try/clearResp
      accum->Nat.Add(Bytes.Length(bytes))
      Nat.Compare(accum, 16_384).case {
        .greater! => {
          yield.item(.chunk(accum) dst)
          let accum = 0
        }
        .equal! => {}
        .less! => {}
      }
      body.loop
    }
    .end! => {}
  }

  file.close.try
  yield.item(.finished dst)
  yield.end!
}

dec ShowDownloadProgress : [List<Event>, Os.Path] !
def ShowDownloadProgress = [events, rootPath] chan exit {
  let getName: box [Os.Path] String = box [path]
    path.absolute
      ->TrimStart(rootPath.absolute)
      ->String.FromBytes

  catch e => {
    Debug.Log(Concat(*("Stdout error: ", e)))
    progress.close
    exit!
  }

  let stdout = Os.Stdout
  let progress = NewDownloadProgress

  events.begin.case {
    .item(event) => {
      event.case {
        .started(expectedSize) path => {
          let name = getName(path)
          progress.addFile(name, expectedSize)
          stdout.write(Concat(*("\nStarted downloading ", name, "\n"))).try
        }

        .chunk(size) path => {
          let name = getName(path)
          progress.addChunk(name, size)
          progress.getStats[numFiles, sumDownloaded, sumExpected]
          stdout.write(Concat(*(
            "\r", Int.ToString(numFiles), " file(s) in progress: ",
            Int.ToString(sumDownloaded), "B / ", Int.ToString(sumExpected), "B",
          ))).try
        }

        .finished path => {
          let name = getName(path)
          progress.removeFile(name)
          stdout.write(Concat(*("\nFinished downloading ", name, "\n"))).try
        }

        .failed(msg) path => {
          let name = getName(path)
          progress.removeFile(name)
          stdout.write(Concat(*("\nFailed to download ", name, ": ", msg, "\n"))).try
        }
      }
      stdout.flush.try
      events.loop
    }
    .end! => {}
  }

  progress.close
  stdout.close
  exit!
}

type DownloadProgress = iterative choice {
  .close => !,
  .getStats => (Nat, Int, Int) self,
  .addFile(String, Nat) => self,
  .removeFile(String) => self,
  .addChunk(String, Nat) => self,
}

def NewDownloadProgress: DownloadProgress = do {
  let filesInProgress = Map.String(type (Nat) Nat)(*())
  let sumDownloaded: Int = 0
  let sumExpected: Int = 0
} in begin case {
  .close => let _ = filesInProgress.list in !,

  .getStats => do {
    filesInProgress.size[numFiles]
  } in (numFiles, sumDownloaded, sumExpected) loop,

  .addFile(name, expected) => do {
    filesInProgress.entry(name)[default((0) 0) (oldExpected) downloaded]
    filesInProgress.put((expected) downloaded)
    sumExpected->Int.Sub(oldExpected)
    sumExpected->Int.Add(expected)
  } in loop,

  .removeFile(name) => do {
    filesInProgress.entry(name)[default((0) 0) (expected) downloaded]
    filesInProgress.delete
    sumExpected->Int.Sub(expected)
    sumDownloaded->Int.Sub(downloaded)
  } in loop,

  .addChunk(name, size) => do {
    filesInProgress.entry(name)[default((0) 0) (expected) downloaded]
    filesInProgress.put((expected) Nat.Add(downloaded, size))
    sumDownloaded->Int.Add(size)
  } in loop,
}

// ---

dec FanIn : [type a] [List<List<a>>] List<a>
def FanIn = [type a] [lists] chan yield {
  let yield = Cell.Share(type dual List<a>)(yield, chan cell {
    lists.begin.case {
      .item(list) => {
        cell.split(chan cell {
          list.begin.case {
            .item(value) => {
              cell.take[yield]
              yield.item(value)
              cell.put(yield)
              list.loop
            }
            .end! => { cell.end! }
          }
        })
        lists.loop
      }
      .end! => { cell.end! }
    }
  })
  yield.end!
}

dec Map : [type a, b] [List<a>, box [a] b] List<b>
def Map = [type a, b] [list, f] list.begin.case {
  .end! => .end!,
  .item(x) xs => .item(f(x)) xs.loop,
}

dec Concat : [List<String>] String
def Concat = [strings] do {
  let builder = String.Builder
  strings.begin.case {
    .item(string) => {
      builder.add(string)
      strings.loop
    }
    .end! => {}
  }
} in builder.build

dec TrimStart : [Bytes, Bytes] Bytes
def TrimStart = [whole, start]
  Bytes.Parser(whole).matchEnd(.bytes start, .repeat.one.any!).case {
    .end _ => whole,
    .fail p => do { p.close } in whole,
    .match(_, trimmed)! => trimmed,
  }