Skip to content
scalive
Menu
ConnectingLiveReconnectingOffline

File uploads

Prerequisites

This guide builds on:

Choose A Destination And Set Hard Limits

Start with the resource boundary, not the file input. A LiveUploadDef[Result] fixes the upload name, selection policy, limits, destination result type, and whether transfer starts automatically:

scala
private val TextFiles = LiveUploadDef.inMemory(
  name = "text-file",
  accept = LiveUploadAccept.only(".txt", ".md"),
  maxEntries = 1,
  maxFileSize = 64L * 1024L
)

Choose the factory by where bytes should travel:

  • inMemory buffers every byte in server heap and returns Chunk[Byte]. Use it only for small, strictly bounded uploads with modest concurrency.

  • hosted sends browser chunks through Scalive to a LiveUploadWriter[State, Result]. Use it to stream into a temporary file, scanner, object-store SDK, or another application-managed sink without collecting the whole file in heap.

  • external asks a LiveUploadExternalUploader[Result] for browser-visible configuration, then lets the browser send bytes directly to another service. Scalive retains only the server-side Result handle.

The browser reports the name, media type, size, relative path, modification time, and progress. Treat all of them as untrusted. LiveUploadAccept improves selection and performs early preflight checks; it does not prove the file's type, contents, destination size, or integrity.

Allow The Upload During Every Mount

Keep one stable definition value and retain the returned LiveUpload[Result] snapshot:

scala
def mount(ctx: MountContext): Task[Model] =
  ctx.uploads.allow(TextFiles).map(Model(_))

Disconnected and connected mounts own independent registrations. Calling allow in mount supports both the initial HTML and connected socket without sharing state between visitors.

An upload and each LiveUploadEntry are immutable point-in-time views. Progress, cancellation, and consumption return replacement snapshots. Store the returned value or refresh it through ctx.uploads.get; rendering an old value renders old protocol attributes and status.

Render Selection, Drop, And Progress Controls

Use liveFileInput rather than assembling protocol attributes by hand. Put upload.dropTarget on any element that should accept files dropped for this upload:

scala
form(
  on.change(_ => Msg.Validate),
  on.submit(Msg.Save),
  div(
    cls := "drop-zone",
    model.upload.dropTarget,
    liveFileInput(
      model.upload,
      aria.label := "Text files",
      model.upload.onProgress(Msg.Progress)
    )
  )
)

The file input supplies the upload reference, active, preflighted, and completed entry references, accepted values, multiplicity, automatic-upload marker, and Phoenix upload hook. A drop target only routes dropped files to that input; it does not relax acceptance, count, or size checks.

upload.onProgress(Msg.Progress) is a DOM event binding. Handle the message by fetching the latest snapshot rather than trusting raw browser metadata:

scala
case Msg.Progress =>
  ctx.uploads.get(TextFiles).map(_.fold(model)(upload => model.copy(upload = upload)))

The overload taking Map[String, String] => Msg exposes raw ref, entry_ref, progress, and optional error values. They are client-controlled protocol data, not authorization evidence.

Render upload-wide and entry-specific uploadErrors explicitly. Display entry.progress as status only; 100 percent is not proof that valid content reached durable storage.

Choose Manual Or Automatic Transfer

The default autoUpload = false waits for the form submit before valid entries start transferring. Set autoUpload = true when selection should start transfer immediately:

scala
private val Avatars = LiveUploadDef.hosted(
  name = "avatar",
  accept = LiveUploadAccept.only("image/jpeg", "image/png"),
  writer = avatarWriter,
  maxFileSize = 2L * 1024L * 1024L,
  autoUpload = true
)

Automatic transfer does not consume an entry, publish a result, or submit the surrounding form. Keep a separate Save action that validates application state and calls consume or consumeCompleted after entries complete.

For a server-side progress hook, pass progress = Some(...) on the definition:

scala
val progress = new LiveUploadProgress[StoredTempFile]:
  def onProgress(entry: LiveUploadEntry[StoredTempFile]): Task[Unit] =
    metrics.record(entry.ref, entry.progress)

This callback runs after an accepted browser progress report has updated runtime state. It is not called for every hosted chunk, and its effect failing fails the progress operation without rolling the state update back. Use it for lightweight observation or orchestration, not exact byte accounting. The writer is the authoritative place to count hosted bytes.

Stream Through A Hosted Writer

A hosted writer owns an in-progress State and eventually produces a typed Result. A temporary-file writer can follow this shape:

scala
final case class PendingFile(path: Path, expected: Long, written: Long)
final case class StoredTempFile(path: Path, bytes: Long)

val MaxBytes = 2L * 1024L * 1024L

val writer = new LiveUploadWriter[PendingFile, StoredTempFile]:
  def init(client: UploadClientMetadata): Task[PendingFile] =
    ZIO.attemptBlocking {
      if client.sizeBytes > MaxBytes then
        throw new IllegalArgumentException("upload exceeds destination limit")
      val path = Files.createTempFile("scalive-upload-", ".pending")
      PendingFile(path, client.sizeBytes, 0L)
    }

  def writeChunk(data: Chunk[Byte], state: PendingFile): Task[PendingFile] =
    ZIO.attemptBlocking {
      val nextSize = state.written + data.length
      if nextSize > MaxBytes then
        throw new IllegalArgumentException("upload exceeds destination limit")
      Files.write(state.path, data.toArray, StandardOpenOption.APPEND)
      state.copy(written = nextSize)
    }

  def complete(state: PendingFile): Task[StoredTempFile] =
    ZIO.attempt {
      if state.written != state.expected then
        throw new IllegalStateException("upload length mismatch")
      StoredTempFile(state.path, state.written)
    }

  def abort(state: PendingFile, reason: LiveUploadAbortReason): Task[Unit] =
    ZIO.attemptBlocking(Files.deleteIfExists(state.path)).unit

  def discard(result: StoredTempFile): Task[Unit] =
    ZIO.attemptBlocking(Files.deleteIfExists(result.path)).unit

Use a server-generated path. Never resolve client.fileName or client.relativePath directly into storage. Enforce a destination-side byte limit and inspect or scan content in complete or before consumption. Chunk boundaries are transport details and are not content boundaries.

The runtime threads only successfully returned state. abort releases an initialized state after cancellation, disallow, component removal, socket shutdown, or upload failure. discard releases a completed result still owned by the runtime. If a failed init, writeChunk, or complete call creates a side effect not represented by the last successful state, that call must clean it itself. Framework cleanup is best-effort and cannot run after process death, so also sweep abandoned temporary resources by age.

Upload Directly To An External Service

An external uploader authorizes one entry, reserves a server-side handle, and returns only browser-safe configuration:

scala
final case class ReservedObject(key: String, uploadId: String)

val uploader = new LiveUploadExternalUploader[ReservedObject]:
  def preflight(client: UploadClientMetadata): Task[LiveExternalUploadResult[ReservedObject]] =
    authorizeUpload(client) *> objectStore.preparePut(client.sizeBytes).map { prepared =>
      val config = ExternalUploadClientConfig(Json.Obj(
        "uploader" -> Json.Str("object-store"),
        "url"      -> Json.Str(prepared.signedUrl),
        "method"   -> Json.Str("PUT")
      ))
      LiveExternalUploadResult.Ready(config, ReservedObject(prepared.key, prepared.uploadId))
    }

  override def discard(result: ReservedObject): Task[Unit] =
    objectStore.abort(result.key, result.uploadId)

Authorize in preflight, generate the object key on the server, scope signed credentials narrowly, and keep server secrets out of ExternalUploadClientConfig. Return LiveExternalUploadResult.Error(meta) for a safe structured rejection. If preparation fails or rejects after reserving a resource but before returning Ready, release that resource in preflight; Scalive has no result to pass to discard yet.

The uploader string must match a Phoenix external uploader installed when the browser creates LiveSocket. For a signed PUT workflow:

javascript
const uploaders = {
  "object-store": (entries, onViewError) => {
    entries.forEach(entry => {
      const xhr = new XMLHttpRequest()
      onViewError(() => xhr.abort())
      xhr.upload.addEventListener("progress", event => {
        if (event.lengthComputable) {
          entry.progress(Math.round((event.loaded / event.total) * 100))
        }
      })
      xhr.addEventListener("load", () => {
        if (xhr.status >= 200 && xhr.status < 300) entry.progress(100)
        else entry.error()
      })
      xhr.addEventListener("error", () => entry.error())
      xhr.open(entry.meta.method, entry.meta.url, true)
      xhr.send(entry.file)
    })
  },
}

const liveSocket = new LiveSocket("/live", Socket, { params, uploaders })

Scalive never receives external bytes. Browser-reported completion merely makes the prepared result consumable. In the consume callback, query the external service and verify ownership, final size, media type, checksum or integrity, and scan status as appropriate before publishing the object and returning Consume.

Consume And Transfer Ownership

consumeCompleted visits valid completed entries in selection order. It fails while a valid entry is still in progress, skips invalid entries, and is not transactional: an earlier consumed entry stays consumed if a later callback fails.

The framework owns a completed destination result while the callback runs. The returned ConsumeDecision determines what happens next:

  • Consume(value) removes the entry and transfers responsibility for its result to application code. Destination cleanup is not called.

  • Postpone(value) keeps the entry and result framework-owned so a later attempt, cancellation, disallow, component removal, or socket shutdown can release it.

Persist, verify, or atomically move a hosted result before returning Consume. For an external result, verify and finalize the reserved object first. If a retryable application operation fails, return Postpone so the runtime still owns cleanup:

scala
ctx.uploads.consumeCompleted(TextFiles) { completed =>
  repository.publish(completed).foldZIO(
    _ => ZIO.succeed(ConsumeDecision.Postpone(false)),
    _ => ZIO.succeed(ConsumeDecision.Consume(true))
  )
}

Cancel with the current entry snapshot and retain the returned upload:

scala
case Msg.Cancel(entry) =>
  ctx.uploads.cancel(entry).map(upload => model.copy(upload = upload))

A stale or already removed entry fails with a typed upload operation error.

Reset And Release Resources

Disallowing a definition releases every result the runtime still owns. A full reset can then allow the stable definition again:

scala
case Msg.Reset =>
  ctx.uploads.disallow(TextFiles) *>
    ctx.uploads.allow(TextFiles).map(Model(_))

Socket shutdown and component removal also clean framework-owned resources on a best-effort basis. Once Consume transfers ownership, application retention, deletion, and failure compensation policies apply instead.

The complete bounded in-memory implementation is extracted from executable source:

Source
scala
final class TextUploadExample extends LiveView[TextUploadExample.Msg, TextUploadExample.Model]:
  import TextUploadExample.*

  def mount(ctx: MountContext): Task[Model] =
    ctx.uploads.allow(TextFiles).map(Model(_))

  def handleMessage(model: Model, ctx: MessageContext) =
    case Msg.Validate | Msg.Progress => refresh(model, ctx.uploads)
    case Msg.Cancel(entry)           =>
      ctx.uploads.cancel(entry).map(upload => model.copy(upload = upload, notice = None))
    case Msg.Summarize => summarizeCompleted(model, ctx.uploads)
    case Msg.Reset     =>
      ctx.uploads.disallow(TextFiles) *>
        ctx.uploads.allow(TextFiles).map(Model(_))

  override def view(model: Signal[Model]): HtmlElement[Msg] =
    val upload = model.map(_.upload)
    div(
      cls := "docs-text-upload",
      p(
        cls := "docs-example-lede",
        "Choose one small text file. The server validates UTF-8, keeps only aggregate facts, and immediately releases the uploaded bytes."
      ),
      model
        .map(_.notice).option(notice => p(role := "status", cls := "docs-upload-notice", notice)),
      form(
        dataAttr("text-upload-form") := "",
        on.change(_ => Msg.Validate),
        on.submit(Msg.Summarize),
        div(
          cls := "docs-upload-dropzone",
          upload.dropTarget,
          label(forId := upload.map(_.ref.value), "Text file"),
          liveFileInput(
            upload,
            dataAttr("text-upload-input") := "",
            upload.onProgress(Msg.Progress)
          ),
          p(cls := "docs-upload-help", "Accepted: .txt and .md. Maximum: one file, 64 KiB."),
          errorList(uploadErrors(upload)),
          div(
            cls := "docs-upload-entries",
            upload.map(_.entries).splitBy(_.ref) { (_, entry) =>
              articleTag(
                cls := "docs-upload-entry",
                div(
                  strong(entry.map(_.client.fileName)),
                  span(entry.map(entry => formatBytes(entry.client.sizeBytes)))
                ),
                progressTag(
                  value      := entry.map(_.progress.toString),
                  maxAttr    := "100",
                  aria.label := entry.map(entry => s"Upload progress for ${entry.client.fileName}")
                ),
                span(
                  entry.map(entry =>
                    if entry.status == LiveUploadEntryStatus.Completed then "Ready"
                    else s"${entry.progress}%"
                  )
                ),
                button(typ := "button", on.click(entry.map(Msg.Cancel(_))), "Cancel"),
                errorList(uploadErrors(entry))
              )
            }
          )
        ),
        div(
          cls := "docs-upload-actions",
          button(
            typ := "submit",
            submission.replaceTextWith("Summarizing..."),
            "Summarize completed file"
          ),
          button(typ := "button", on.click(Msg.Reset), "Reset upload")
        )
      ),
      div(
        cls := "docs-upload-summaries",
        model.map(_.summaries).splitBy(_.id) { (_, summary) =>
          articleTag(
            dataAttr("upload-summary") := "",
            cls                        := "docs-upload-summary",
            h3(dataAttr("summary-name") := "", summary.map(_.fileName)),
            dl(
              div(
                dt("Size"),
                dd(dataAttr("summary-bytes") := "", summary.map(value => formatBytes(value.bytes)))
              ),
              div(
                dt("Lines"),
                dd(
                  dataAttr("summary-lines") := "",
                  summary.map(value => plural(value.lines, "line"))
                )
              ),
              div(
                dt("Words"),
                dd(
                  dataAttr("summary-words") := "",
                  summary.map(value => plural(value.words, "word"))
                )
              )
            )
          )
        }
      )
    )
  end view

  private def refresh(model: Model, uploads: Uploads): Task[Model] =
    uploads.get(TextFiles).map(_.fold(model)(upload => model.copy(upload = upload, notice = None)))

  private def summarizeCompleted(model: Model, uploads: Uploads): Task[Model] =
    uploads
      .consumeCompleted(TextFiles) { completed =>
        ZIO.succeed(ConsumeDecision.Consume(summarize(completed)))
      }.map { case (results, upload) =>
        val summaries = results.collect { case Right(summary) => summary }
        val rejected  = results.count(_.isLeft)
        val notice    =
          if rejected > 0 then Some("The file was discarded because it was not valid UTF-8 text.")
          else if summaries.isEmpty then
            Some("Finish uploading a valid file before summarizing it.")
          else Some("Summary created. The uploaded bytes were discarded.")
        model.copy(upload = upload, summaries = model.summaries ++ summaries, notice = notice)
      }

  private def summarize(completed: CompletedUpload[Chunk[Byte]]): Either[Unit, Summary] =
    decodeUtf8(completed.result).map { text =>
      val lines = if text.isEmpty then 0 else text.linesIterator.size
      val words = Word.findAllIn(text).size
      Summary(
        id = completed.ref.value,
        fileName = completed.client.fileName,
        bytes = completed.result.length.toLong,
        lines = lines,
        words = words
      )
    }

  private def decodeUtf8(bytes: Chunk[Byte]): Either[Unit, String] =
    val decoder = StandardCharsets.UTF_8
      .newDecoder()
      .onMalformedInput(CodingErrorAction.REPORT)
      .onUnmappableCharacter(CodingErrorAction.REPORT)
    try Right(decoder.decode(ByteBuffer.wrap(bytes.toArray)).toString)
    catch case _: java.nio.charset.CharacterCodingException => Left(())

  private def errorList(errors: Signal[List[LiveUploadError]]): HtmlElement[Nothing] =
    div(
      errors.map(_.distinct).splitBy(identity) { (_, error) =>
        p(role := "alert", error.map(uploadErrorMessage))
      }
    )
end TextUploadExample

object TextUploadExample:
  private val MaxFileSize = 64L * 1024L
  private val Word        = raw"\S+".r

  private val TextFiles = LiveUploadDef.inMemory(
    name = "text-file",
    accept = LiveUploadAccept.only(".txt", ".md"),
    maxEntries = 1,
    maxFileSize = MaxFileSize
  )

  final case class Summary(id: String, fileName: String, bytes: Long, lines: Int, words: Int)

  final case class Model(
    upload: LiveUpload[Chunk[Byte]],
    summaries: Vector[Summary] = Vector.empty,
    notice: Option[String] = None)

  enum Msg:
    case Validate
    case Progress
    case Cancel(entry: LiveUploadEntry[Chunk[Byte]])
    case Summarize
    case Reset

  private def uploadErrorMessage(error: LiveUploadError): String = error match
    case LiveUploadError.TooManyFiles => "Choose one file at a time."
    case LiveUploadError.NotAccepted  => "Choose a .txt or .md file."
    case LiveUploadError.TooLarge     => "Choose a file no larger than 64 KiB."
    case _                            => "The upload failed. Remove the file and try again."

  private def formatBytes(bytes: Long): String =
    if bytes < 1024L then s"$bytes B" else f"${bytes.toDouble / 1024.0}%.1f KiB"

  private def plural(value: Int, unit: String): String =
    s"$value $unit${if value == 1 then "" else "s"}"
end TextUploadExample

View source (documentation/site/src/scalive/docs/examples/TextUploadExample.scala:11-203)

Try the summarize-and-discard text upload. Its X-ray shows lifecycle state and aggregate counts; upload chunks are represented only by their byte length, never their contents.