Uploading directories - first draft

Author: marcinczenko

codex/blockexchange/engine/advertiser.nim

@@ -25,6 +25,9 @@ import ../../stores/blockstore
 import ../../logutils
 import ../../manifest
 
+# tarballs
+import ../../tarballs/[directorymanifest, decoding]
+
 logScope:
   topics = "codex discoveryengine advertiser"
 
@@ -66,7 +69,11 @@ proc advertiseBlock(b: Advertiser, cid: Cid) {.async: (raises: [CancelledError])
         return
 
       without manifest =? Manifest.decode(blk), err:
-        error "Unable to decode as manifest", err = err.msg
+        # Try if it not a directory manifest
+        without manifest =? DirectoryManifest.decode(blk), err:
+          error "Unable to decode as manifest", err = err.msg
+          return
+        await b.addCidToQueue(cid)
         return
 
       # announce manifest cid and tree cid

codex/node.nim

@@ -13,6 +13,7 @@ import std/options
 import std/sequtils
 import std/strformat
 import std/sugar
+import std/streams
 import times
 
 import pkg/taskpools
@@ -47,6 +48,9 @@ import ./logutils
 import ./utils/asynciter
 import ./utils/trackedfutures
 
+# tarball
+import ./tarballs/[tarballs, encoding, decoding, stdstreamwrapper, directorymanifest]
+
 export logutils
 
 logScope:
@@ -132,6 +136,31 @@ proc fetchManifest*(self: CodexNodeRef, cid: Cid): Future[?!Manifest] {.async.}
 
   return manifest.success
 
+proc fetchDirectoryManifest*(
+    self: CodexNodeRef, cid: Cid
+): Future[?!DirectoryManifest] {.async.} =
+  ## Fetch and decode a manifest block
+  ##
+
+  if err =? cid.isManifest.errorOption:
+    return failure "CID has invalid content type for manifest {$cid}"
+
+  trace "Retrieving manifest for cid", cid
+
+  without blk =? await self.networkStore.getBlock(BlockAddress.init(cid)), err:
+    trace "Error retrieve manifest block", cid, err = err.msg
+    return failure err
+
+  trace "Decoding manifest for cid", cid
+
+  without manifest =? DirectoryManifest.decode(blk), err:
+    trace "Unable to decode as manifest", err = err.msg
+    return failure("Unable to decode as manifest")
+
+  trace "Decoded manifest", cid
+
+  manifest.success
+
 proc findPeer*(self: CodexNodeRef, peerId: PeerId): Future[?PeerRecord] {.async.} =
   ## Find peer using the discovery service from the given CodexNode
   ##
@@ -486,6 +515,83 @@ proc iterateManifests*(self: CodexNodeRef, onManifest: OnManifest) {.async.} =
 
       onManifest(cid, manifest)
 
+proc storeDirectoryManifest*(
+    self: CodexNodeRef, manifest: DirectoryManifest
+): Future[?!bt.Block] {.async.} =
+  let encodedManifest = manifest.encode()
+
+  without blk =? bt.Block.new(data = encodedManifest, codec = ManifestCodec), error:
+    trace "Unable to create block from manifest"
+    return failure(error)
+
+  if err =? (await self.networkStore.putBlock(blk)).errorOption:
+    trace "Unable to store manifest block", cid = blk.cid, err = err.msg
+    return failure(err)
+
+  success blk
+
+proc storeTarball*(
+    self: CodexNodeRef, stream: AsyncStreamReader
+): Future[?!string] {.async.} =
+  info "Storing tarball data"
+
+  # Just as a proof of concept, we process tar bar in memory
+  # Later to see how to do actual streaming to either store
+  # tarball locally in some tmp folder, or to process the
+  # tarball incrementally 
+  let tarballBytes = await stream.read()
+  let stream = newStringStream(string.fromBytes(tarballBytes))
+
+  proc onProcessedTarFile(
+      stream: Stream, fileName: string
+  ): Future[?!Cid] {.gcsafe, async: (raises: [CancelledError]).} =
+    try:
+      echo "onProcessedTarFile:name: ", fileName
+      let stream = newStdStreamWrapper(stream)
+      await self.store(stream, filename = some fileName)
+    except CancelledError as e:
+      raise e
+    except CatchableError as e:
+      error "Error processing tar file", fileName, exc = e.msg
+      return failure(e.msg)
+
+  proc onProcessedTarDir(
+      name: string, cids: seq[Cid]
+  ): Future[?!Cid] {.gcsafe, async: (raises: [CancelledError]).} =
+    try:
+      echo "onProcessedTarDir:name: ", name
+      echo "onProcessedTarDir:cids: ", cids
+      let directoryManifest = newDirectoryManifest(name = name, cids = cids)
+      without manifestBlk =? await self.storeDirectoryManifest(directoryManifest), err:
+        error "Unable to store manifest"
+        return failure(err)
+      manifestBlk.cid.success
+    except CancelledError as e:
+      raise e
+    except CatchableError as e:
+      error "Error processing tar dir", name, exc = e.msg
+      return failure(e.msg)
+
+  let tarball = Tarball()
+  if err =? (await tarball.open(stream, onProcessedTarFile)).errorOption:
+    error "Unable to open tarball", err = err.msg
+    return failure(err)
+  echo "tarball = ", $tarball
+  without root =? tarball.findRootDir(), err:
+    return failure(err.msg)
+  echo "root = ", root
+  let dirs = processDirEntries(tarball)
+  echo "dirs = ", dirs
+  without tree =? (await buildTree(root = root, dirs = dirs, onProcessedTarDir)), err:
+    error "Unable to build tree", err = err.msg
+    return failure(err)
+  echo ""
+  echo "preorderTraversal:"
+  let json = newJArray()
+  preorderTraversal(tree, json)
+  echo "json = ", json
+  success($json)
+
 proc setupRequest(
     self: CodexNodeRef,
     cid: Cid,

codex/rest/api.nim

@@ -39,6 +39,8 @@ import ../manifest
 import ../streams/asyncstreamwrapper
 import ../stores
 import ../utils/options
+# tarballs
+import ../tarballs/directorymanifest
 
 import ./coders
 import ./json
@@ -52,6 +54,11 @@ declareCounter(codex_api_downloads, "codex API downloads")
 proc validate(pattern: string, value: string): int {.gcsafe, raises: [Defect].} =
   0
 
+proc formatDirectoryManifest(
+    cid: Cid, manifest: DirectoryManifest
+): RestDirectoryContent =
+  return RestDirectoryContent.init(cid, manifest)
+
 proc formatManifest(cid: Cid, manifest: Manifest): RestContent =
   return RestContent.init(cid, manifest)
 
@@ -179,7 +186,76 @@ proc getFilenameFromContentDisposition(contentDisposition: string): ?string =
   return filename[0 ..^ 2].some
 
 proc initDataApi(node: CodexNodeRef, repoStore: RepoStore, router: var RestRouter) =
-  let allowedOrigin = router.allowedOrigin # prevents capture inside of api defintion
+  let allowedOrigin = router.allowedOrigin # prevents capture inside of api definition
+
+  router.api(MethodOptions, "/api/codex/v1/tar") do(
+    resp: HttpResponseRef
+  ) -> RestApiResponse:
+    if corsOrigin =? allowedOrigin:
+      resp.setCorsHeaders("POST", corsOrigin)
+      resp.setHeader(
+        "Access-Control-Allow-Headers", "content-type, content-disposition"
+      )
+
+    resp.status = Http204
+    await resp.sendBody("")
+
+  router.rawApi(MethodPost, "/api/codex/v1/tar") do() -> RestApiResponse:
+    ## Upload a file in a streaming manner
+    ##
+
+    trace "Handling upload of a tar file"
+    var bodyReader = request.getBodyReader()
+    if bodyReader.isErr():
+      return RestApiResponse.error(Http500, msg = bodyReader.error())
+
+    # Attempt to handle `Expect` header
+    # some clients (curl), wait 1000ms
+    # before giving up
+    #
+    await request.handleExpect()
+
+    var mimetype = request.headers.getString(ContentTypeHeader).some
+
+    if mimetype.get() != "":
+      let mimetypeVal = mimetype.get()
+      var m = newMimetypes()
+      let extension = m.getExt(mimetypeVal, "")
+      if extension == "":
+        return RestApiResponse.error(
+          Http422, "The MIME type '" & mimetypeVal & "' is not valid."
+        )
+    else:
+      mimetype = string.none
+
+    const ContentDispositionHeader = "Content-Disposition"
+    let contentDisposition = request.headers.getString(ContentDispositionHeader)
+    let filename = getFilenameFromContentDisposition(contentDisposition)
+
+    if filename.isSome and not isValidFilename(filename.get()):
+      return RestApiResponse.error(Http422, "The filename is not valid.")
+
+    # Here we could check if the extension matches the filename if needed
+
+    let reader = bodyReader.get()
+    let stream = AsyncStreamReader(reader)
+
+    try:
+      without json =? (await node.storeTarball(stream = stream)), error:
+        error "Error uploading tarball", exc = error.msg
+        return RestApiResponse.error(Http500, error.msg)
+
+      codex_api_uploads.inc()
+      trace "Uploaded tarball", result = json
+      return RestApiResponse.response(json, contentType = "application/json")
+    except CancelledError:
+      trace "Upload cancelled error"
+      return RestApiResponse.error(Http500)
+    except AsyncStreamError:
+      trace "Async stream error"
+      return RestApiResponse.error(Http500)
+    finally:
+      await reader.closeWait()
 
   router.api(MethodOptions, "/api/codex/v1/data") do(
     resp: HttpResponseRef
@@ -363,6 +439,24 @@ proc initDataApi(node: CodexNodeRef, repoStore: RepoStore, router: var RestRoute
     let json = %formatManifest(cid.get(), manifest)
     return RestApiResponse.response($json, contentType = "application/json")
 
+  router.api(MethodGet, "/api/codex/v1/data/{cid}/network/dirmanifest") do(
+    cid: Cid, resp: HttpResponseRef
+  ) -> RestApiResponse:
+    ## Download only the directory manifest.
+    ##
+
+    var headers = buildCorsHeaders("GET", allowedOrigin)
+
+    if cid.isErr:
+      return RestApiResponse.error(Http400, $cid.error(), headers = headers)
+
+    without manifest =? (await node.fetchDirectoryManifest(cid.get())), err:
+      error "Failed to fetch directory manifest", err = err.msg
+      return RestApiResponse.error(Http404, err.msg, headers = headers)
+
+    let json = %formatDirectoryManifest(cid.get(), manifest)
+    return RestApiResponse.response($json, contentType = "application/json")
+
   router.api(MethodGet, "/api/codex/v1/space") do() -> RestApiResponse:
     let json =
       %RestRepoStore(

codex/rest/json.nim

@@ -9,6 +9,8 @@ import ../utils/json
 import ../manifest
 import ../units
 
+import ../tarballs/directorymanifest
+
 export json
 
 type
@@ -47,6 +49,10 @@ type
     cid* {.serialize.}: Cid
     manifest* {.serialize.}: Manifest
 
+  RestDirectoryContent* = object
+    cid* {.serialize.}: Cid
+    manifest* {.serialize.}: DirectoryManifest
+
   RestContentList* = object
     content* {.serialize.}: seq[RestContent]
 
@@ -81,6 +87,11 @@ proc init*(_: type RestContentList, content: seq[RestContent]): RestContentList
 proc init*(_: type RestContent, cid: Cid, manifest: Manifest): RestContent =
   RestContent(cid: cid, manifest: manifest)
 
+proc init*(
+    _: type RestDirectoryContent, cid: Cid, manifest: DirectoryManifest
+): RestDirectoryContent =
+  RestDirectoryContent(cid: cid, manifest: manifest)
+
 proc init*(_: type RestNode, node: dn.Node): RestNode =
   RestNode(
     nodeId: RestNodeId.init(node.id),

codex/tarballs/decoding.nim

@@ -0,0 +1,60 @@
+## Nim-Codex
+## Copyright (c) 2021 Status Research & Development GmbH
+## Licensed under either of
+##  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+##  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+## at your option.
+## This file may not be copied, modified, or distributed except according to
+## those terms.
+
+{.push raises: [].}
+
+import pkg/libp2p/cid
+import pkg/libp2p/multihash
+import pkg/libp2p/protobuf/minprotobuf
+
+import pkg/questionable/results
+
+import ../blocktype
+import ./directorymanifest
+
+func decode*(_: type DirectoryManifest, data: openArray[byte]): ?!DirectoryManifest =
+  # ```protobuf
+  #   Message DirectoryManifest {
+  #     Message Cid {
+  #       bytes data = 1;
+  #     }
+  #     string name = 1;
+  #     repeated Cid cids = 2;
+  # ```
+
+  var
+    pbNode = initProtoBuffer(data)
+    pbInfo: ProtoBuffer
+    name: string
+    cids: seq[Cid]
+    cidsBytes: seq[seq[byte]]
+
+  if pbNode.getField(1, name).isErr:
+    return failure("Unable to decode `name` from DirectoryManifest")
+
+  if ?pbNode.getRepeatedField(2, cidsBytes).mapFailure:
+    for cidEntry in cidsBytes:
+      var pbCid = initProtoBuffer(cidEntry)
+      var dataBuf = newSeq[byte]()
+      if pbCid.getField(1, dataBuf).isErr:
+        return failure("Unable to decode piece `data` to Cid")
+      without cid =? Cid.init(dataBuf).mapFailure, err:
+        return failure(err.msg)
+      cids.add(cid)
+
+  DirectoryManifest(name: name, cids: cids).success
+
+func decode*(_: type DirectoryManifest, blk: Block): ?!DirectoryManifest =
+  ## Decode a directory manifest using `decoder`
+  ##
+
+  if not ?blk.cid.isManifest:
+    return failure "Cid is not a Directory Manifest Cid"
+
+  DirectoryManifest.decode(blk.data)