tricu

artifacts.tri (4987B)

---

 !import base as B
 !import native_math as N
 !import http as HTTP
 !import io as IO
 !import io_storage as Storage
 !import handlers as H
 !import arboricx_server as Store
 !export write
 !export publish
 !export publishWith
 
 -- Source-owned upload policy. Only writes observed by this boundary advance
 -- its byte count; producer acknowledgements report producer status.
 write bytes = H.request "artifact-stream" "write" bytes
 onOk action next = H.bind (H.liftIO action) (reply : case reply of
   ((t t), (value, t)) = next value
   (t, (_, t)) = H.pure reply
   _ = H.fail "invalid-response" "artifact-stream" "host" "Invalid host Result")
 
 hashReader reader digest remaining = IO.onOk (IO.readSome reader 4096) (reply : case reply of
   ("eof", t) = case remaining of
     0 = IO.finishDigest digest
     _ = IO.pure (B.err "Existing artifact length mismatch" t)
   ("data", bytes) = let count = HTTP.httpBodyLength bytes in case N.lte count remaining of
     (t t) = IO.onOk (IO.updateDigest digest bytes)
       (_ : hashReader reader digest (N.subtract remaining count))
     _ = IO.pure (B.err "Existing artifact length mismatch" t)
   _ = IO.fail "Invalid read reply")
 
 -- Accept an existing name only after verifying its complete contents and length
 -- independently with fresh reader/digest resources.
 verifyWinner destination expected wanted = IO.scope (owner : do IO.onOk
   reader <- IO.openReader owner destination
   digest <- IO.openDigest owner "sha256"
   actual <- hashReader reader digest expected
   case B.equal? actual wanted of
     (t t) = do IO.onOk
       IO.syncFile reader
       IO.syncDirectory (Storage.parentPath destination)
       IO.pure (B.ok t t)
     _ = IO.pure (B.err "Existing artifact digest mismatch" t))
 
 -- Destination policy runs only after exact-length completion and sealing. It
 -- returns a handler Result path and can create a digest-derived shard. Native
 -- publication remains generic and rejects cross-filesystem moves without copying.
 finish destination expected stage digest = onOk (IO.finishDigest digest) (hash :
   case B.checkContract Store.bundleHashC hash of
     ((t t), (checked, t)) = onOk (IO.sealStage stage) (_ :
       H.bind (destination checked) (reply : case reply of
         ((t t), (path, t)) = onOk (IO.publishNew stage path) (outcome : case outcome of
           "published" = H.pure (B.ok checked t)
           "exists" = onOk (verifyWinner path expected checked) (_ : H.pure (B.ok checked t))
           _ = H.fail "invalid-response" "artifact-stream" "publish" "Invalid publication reply")
         (t, (_, t)) = H.pure reply
         _ = H.fail "invalid-response" "artifact-stream" "destination" "Expected destination Result"))
     _ = H.fail "invalid-response" "artifact-stream" "digest" "Invalid digest reply")
 
 publish directory expected producer = publishWith directory expected
   (digest : H.pure (B.ok (B.append directory (B.pair 47 digest)) t)) producer
 publishWith directory expected destination producer = case B.checkContract B.nat? expected of
   ((t t), (count, t)) = case N.lte count 67108864 of
     (t t) = H.scope (owner : onOk (IO.openStage owner directory) (stage :
       onOk (IO.openDigest owner "sha256") (digest : let rec
         walk current remaining = case current of
           (100, reply) = case reply of
             ((t t), (t, t)) = case remaining of
               0 = finish destination count stage digest
               _ = H.pure (B.err "Incomplete artifact stream" t)
             (t, (_, t)) = H.pure reply
             _ = H.fail "invalid-response" "artifact-stream" "producer" "Expected Result unit"
           (103, [kind service operation detail labels]) = B.pair 105
             (B.pair labels (H.fail kind service operation detail))
           (106, [["artifact-stream" "write" bytes] labels next]) =
             onOk (IO.writeAll stage bytes) (_ : case bytes of
               [] = H.pure (B.err "Empty artifact chunk" t)
               _ = let length = HTTP.httpBodyLength bytes in case N.lte length remaining of
                 (t t) = onOk (IO.updateDigest digest bytes)
                   (_ : walk (next (B.ok t t)) (N.subtract remaining length))
                 _ = H.pure (B.err "Artifact exceeds declared length" t))
           (106, [[service operation argument] labels next]) = H.bind
             (B.pair 105 (B.pair labels (H.request service operation argument)))
             (reply : walk (next reply) remaining)
           (107, [labels body next]) = B.pair 105 (B.pair labels
             (H.fail "producer-boundary" "artifact-stream" "publish" "Producer must be sequential"))
           (108, [supervised keep handles labels body next]) = B.pair 105 (B.pair labels
             (H.fail "producer-boundary" "artifact-stream" "publish" "Producer must be sequential"))
           _ = H.fail "invalid-view" "artifact-stream" "publish" "Invalid producer view"
         in walk (H.view producer) count)))
     _ = H.pure (B.err "Artifact length exceeds limit" t)
   _ = H.pure (B.err "Invalid artifact length" t)