[skip-ci] initial support for stream installs, add ftp installs.

do NOT build or release binaries of this version, it is not complete and there will be dragons.

sphaira/source/ftpsrv_helper.cpp

@@ -12,14 +12,30 @@
 #include <nx/vfs_nx.h>
 #include <nx/utils.h>
 
+namespace sphaira::ftpsrv {
 namespace {
 
+struct InstallSharedData {
+    std::mutex mutex;
+
+    std::deque<std::string> queued_files;
+
+    void* user;
+    OnInstallStart on_start;
+    OnInstallWrite on_write;
+    OnInstallClose on_close;
+
+    bool in_progress;
+    bool enabled;
+};
+
 const char* INI_PATH = "/config/ftpsrv/config.ini";
 FtpSrvConfig g_ftpsrv_config = {0};
 volatile bool g_should_exit = false;
 bool g_is_running{false};
 Thread g_thread;
 std::mutex g_mutex{};
+InstallSharedData g_shared_data{};
 
 void ftp_log_callback(enum FTP_API_LOG_TYPE type, const char* msg) {
     sphaira::App::NotifyFlashLed();
@@ -29,6 +45,235 @@ void ftp_progress_callback(void) {
     sphaira::App::NotifyFlashLed();
 }
 
+const char* SUPPORTED_EXT[] = {
+    ".nsp", ".xci", ".nsz", ".xcz",
+};
+
+struct VfsUserData {
+    char* path;
+    int valid;
+};
+
+// ive given up with good names.
+void on_thing() {
+    log_write("[FTP] doing on_thing\n");
+    std::scoped_lock lock{g_shared_data.mutex};
+    log_write("[FTP] locked on_thing\n");
+
+    if (!g_shared_data.in_progress) {
+        if (!g_shared_data.queued_files.empty()) {
+            log_write("[FTP] pushing new file data\n");
+            if (!g_shared_data.on_start || !g_shared_data.on_start(g_shared_data.user, g_shared_data.queued_files[0].c_str())) {
+                g_shared_data.queued_files.clear();
+            } else {
+                log_write("[FTP] success on new file push\n");
+                g_shared_data.in_progress = true;
+            }
+        }
+    }
+}
+
+int vfs_install_open(void* user, const char* path, enum FtpVfsOpenMode mode) {
+    {
+        std::scoped_lock lock{g_shared_data.mutex};
+        auto data = static_cast<VfsUserData*>(user);
+        data->valid = 0;
+
+        if (mode != FtpVfsOpenMode_WRITE) {
+            errno = EACCES;
+            return -1;
+        }
+
+        if (!g_shared_data.enabled) {
+            errno = EACCES;
+            return -1;
+        }
+
+        const char* ext = strrchr(path, '.');
+        if (!ext) {
+            errno = EACCES;
+            return -1;
+        }
+
+        bool found = false;
+        for (size_t i = 0; i < std::size(SUPPORTED_EXT); i++) {
+            if (!strcasecmp(ext, SUPPORTED_EXT[i])) {
+                found = true;
+                break;
+            }
+        }
+
+        if (!found) {
+            errno = EINVAL;
+            return -1;
+        }
+
+        // check if we already have this file queued.
+        auto it = std::find(g_shared_data.queued_files.cbegin(), g_shared_data.queued_files.cend(), path);
+        if (it != g_shared_data.queued_files.cend()) {
+            errno = EEXIST;
+            return -1;
+        }
+
+        g_shared_data.queued_files.push_back(path);
+        data->path = strdup(path);
+        data->valid = true;
+    }
+
+    on_thing();
+    log_write("[FTP] got file: %s\n", path);
+    return 0;
+}
+
+int vfs_install_read(void* user, void* buf, size_t size) {
+    errno = EACCES;
+    return -1;
+}
+
+int vfs_install_write(void* user, const void* buf, size_t size) {
+    std::scoped_lock lock{g_shared_data.mutex};
+    if (!g_shared_data.enabled) {
+        errno = EACCES;
+        return -1;
+    }
+
+    auto data = static_cast<VfsUserData*>(user);
+    if (!data->valid) {
+        errno = EACCES;
+        return -1;
+    }
+
+    if (!g_shared_data.on_write || !g_shared_data.on_write(g_shared_data.user, buf, size)) {
+        errno = EIO;
+        return -1;
+    }
+
+    return size;
+}
+
+int vfs_install_seek(void* user, const void* buf, size_t size, size_t off) {
+    errno = ESPIPE;
+    return -1;
+}
+
+int vfs_install_isfile_open(void* user) {
+    std::scoped_lock lock{g_shared_data.mutex};
+    auto data = static_cast<VfsUserData*>(user);
+    return data->valid;
+}
+
+int vfs_install_isfile_ready(void* user) {
+    std::scoped_lock lock{g_shared_data.mutex};
+    auto data = static_cast<VfsUserData*>(user);
+    const auto ready = !g_shared_data.queued_files.empty() && data->path == g_shared_data.queued_files[0];
+    return ready;
+}
+
+int vfs_install_close(void* user) {
+    {
+        log_write("[FTP] closing file\n");
+        std::scoped_lock lock{g_shared_data.mutex};
+        auto data = static_cast<VfsUserData*>(user);
+        if (data->valid) {
+            log_write("[FTP] closing valid file\n");
+
+            auto it = std::find(g_shared_data.queued_files.cbegin(), g_shared_data.queued_files.cend(), data->path);
+            if (it != g_shared_data.queued_files.cend()) {
+                if (it == g_shared_data.queued_files.cbegin()) {
+                    log_write("[FTP] closing current file\n");
+                    if (g_shared_data.on_close) {
+                        g_shared_data.on_close(g_shared_data.user);
+                    }
+
+                    g_shared_data.in_progress = false;
+                } else {
+                    log_write("[FTP] closing other file...\n");
+                }
+
+                g_shared_data.queued_files.erase(it);
+            } else {
+                log_write("[FTP] could not find file in queue...\n");
+            }
+
+            if (data->path) {
+                free(data->path);
+            }
+
+            data->valid = 0;
+        }
+
+        memset(data, 0, sizeof(*data));
+    }
+
+    on_thing();
+    return 0;
+}
+
+int vfs_install_opendir(void* user, const char* path) {
+    return 0;
+}
+
+const char* vfs_install_readdir(void* user, void* user_entry) {
+    return NULL;
+}
+
+int vfs_install_dirlstat(void* user, const void* user_entry, const char* path, struct stat* st) {
+    st->st_nlink = 1;
+    st->st_mode = S_IFDIR | S_IRUSR | S_IRGRP | S_IROTH;
+    return 0;
+}
+
+int vfs_install_isdir_open(void* user) {
+    return 1;
+}
+
+int vfs_install_closedir(void* user) {
+    return 0;
+}
+
+int vfs_install_stat(const char* path, struct stat* st) {
+    st->st_nlink = 1;
+    st->st_mode = S_IFDIR | S_IWUSR | S_IWGRP | S_IWOTH;
+    return 0;
+}
+
+int vfs_install_mkdir(const char* path) {
+    return -1;
+}
+
+int vfs_install_unlink(const char* path) {
+    return -1;
+}
+
+int vfs_install_rmdir(const char* path) {
+    return -1;
+}
+
+int vfs_install_rename(const char* src, const char* dst) {
+    return -1;
+}
+
+FtpVfs g_vfs_install = {
+    .open = vfs_install_open,
+    .read = vfs_install_read,
+    .write = vfs_install_write,
+    .seek = vfs_install_seek,
+    .close = vfs_install_close,
+    .isfile_open = vfs_install_isfile_open,
+    .isfile_ready = vfs_install_isfile_ready,
+    .opendir = vfs_install_opendir,
+    .readdir = vfs_install_readdir,
+    .dirlstat = vfs_install_dirlstat,
+    .closedir = vfs_install_closedir,
+    .isdir_open = vfs_install_isdir_open,
+    .stat = vfs_install_stat,
+    .lstat = vfs_install_stat,
+    .mkdir = vfs_install_mkdir,
+    .unlink = vfs_install_unlink,
+    .rmdir = vfs_install_rmdir,
+    .rename = vfs_install_rename,
+};
+
 void loop(void* arg) {
     while (!g_should_exit) {
         ftpsrv_init(&g_ftpsrv_config);
@@ -44,8 +289,6 @@ void loop(void* arg) {
 
 } // namespace
 
-namespace sphaira::ftpsrv {
-
 bool Init() {
     std::scoped_lock lock{g_mutex};
     if (g_is_running) {
@@ -84,6 +327,9 @@ bool Init() {
     mount_bis = ini_getbool("Nx-App", "mount_bis", mount_bis, INI_PATH);
     save_writable = ini_getbool("Nx-App", "save_writable", save_writable, INI_PATH);
 
+    mount_devices = true;
+    g_ftpsrv_config.timeout = 0;
+
     if (!g_ftpsrv_config.port) {
         return false;
     }
@@ -93,7 +339,13 @@ bool Init() {
         g_ftpsrv_config.anon = true;
     }
 
-    vfs_nx_init(mount_devices, save_writable, mount_bis);
+    const VfsNxCustomPath custom = {
+        .name = "install",
+        .user = NULL,
+        .func = &g_vfs_install,
+    };
+
+    vfs_nx_init(&custom, mount_devices, save_writable, mount_bis);
 
     Result rc;
     if (R_FAILED(rc = threadCreate(&g_thread, loop, nullptr, nullptr, 1024*16, 0x2C, 2))) {
@@ -123,6 +375,20 @@ void Exit() {
     fsdev_wrapUnmountAll();
 }
 
+void InitInstallMode(void* user, OnInstallStart on_start, OnInstallWrite on_write, OnInstallClose on_close) {
+    std::scoped_lock lock{g_shared_data.mutex};
+    g_shared_data.user = user;
+    g_shared_data.on_start = on_start;
+    g_shared_data.on_write = on_write;
+    g_shared_data.on_close = on_close;
+    g_shared_data.enabled = true;
+}
+
+void DisableInstallMode() {
+    std::scoped_lock lock{g_shared_data.mutex};
+    g_shared_data.enabled = false;
+}
+
 } // namespace sphaira::ftpsrv
 
 extern "C" {

sphaira/source/ui/menus/ftp_menu.cpp

@@ -0,0 +1,242 @@
+#include "ui/menus/ftp_menu.hpp"
+#include "yati/yati.hpp"
+#include "app.hpp"
+#include "defines.hpp"
+#include "log.hpp"
+#include "ui/nvg_util.hpp"
+#include "i18n.hpp"
+#include "ftpsrv_helper.hpp"
+#include <cstring>
+
+namespace sphaira::ui::menu::ftp {
+namespace {
+
+constexpr u64 MAX_BUFFER_SIZE = 1024*1024*32;
+constexpr u64 SLEEPNS = 1000;
+volatile bool IN_PUSH_THREAD{};
+
+bool OnInstallStart(void* user, const char* path) {
+    auto menu = (Menu*)user;
+    log_write("[INSTALL] inside OnInstallStart()\n");
+
+    for (;;) {
+        mutexLock(&menu->m_mutex);
+        ON_SCOPE_EXIT(mutexUnlock(&menu->m_mutex));
+
+        if (menu->m_state != State::Progress) {
+            break;
+        }
+
+        if (menu->GetToken().stop_requested()) {
+            return false;
+        }
+
+        svcSleepThread(1e+6);
+    }
+
+    log_write("[INSTALL] OnInstallStart() got state: %u\n", (u8)menu->m_state);
+
+    if (menu->m_source) {
+        log_write("[INSTALL] OnInstallStart() we have source\n");
+        for (;;) {
+            mutexLock(&menu->m_source->m_mutex);
+            ON_SCOPE_EXIT(mutexUnlock(&menu->m_source->m_mutex));
+
+            if (!IN_PUSH_THREAD) {
+                break;
+            }
+
+            if (menu->GetToken().stop_requested()) {
+                return false;
+            }
+
+            svcSleepThread(1e+6);
+        }
+
+        log_write("[INSTALL] OnInstallStart() stopped polling source\n");
+    }
+
+    log_write("[INSTALL] OnInstallStart() doing make_shared\n");
+    menu->m_source = std::make_shared<StreamFtp>(path, menu->GetToken());
+
+    mutexLock(&menu->m_mutex);
+    ON_SCOPE_EXIT(mutexUnlock(&menu->m_mutex));
+    menu->m_state = State::Connected;
+    log_write("[INSTALL] OnInstallStart() done make shared\n");
+
+    return true;
+}
+
+bool OnInstallWrite(void* user, const void* buf, size_t size) {
+    auto menu = (Menu*)user;
+
+    return menu->m_source->Push(buf, size);
+}
+
+void OnInstallClose(void* user) {
+    auto menu = (Menu*)user;
+    menu->m_source->Disable();
+}
+
+} // namespace
+
+StreamFtp::StreamFtp(const fs::FsPath& path, std::stop_token token) {
+    m_path = path;
+    m_token = token;
+    m_buffer.reserve(MAX_BUFFER_SIZE);
+    m_active = true;
+}
+
+Result StreamFtp::ReadChunk(void* buf, s64 size, u64* bytes_read) {
+    while (!m_token.stop_requested()) {
+        mutexLock(&m_mutex);
+        ON_SCOPE_EXIT(mutexUnlock(&m_mutex));
+
+        if (m_buffer.empty()) {
+            if (!m_active) {
+                break;
+            }
+
+            svcSleepThread(SLEEPNS);
+        } else {
+            size = std::min<s64>(size, m_buffer.size());
+            std::memcpy(buf, m_buffer.data(), size);
+            m_buffer.erase(m_buffer.begin(), m_buffer.begin() + size);
+            *bytes_read = size;
+            R_SUCCEED();
+        }
+    }
+
+    return 0x1;
+}
+
+bool StreamFtp::Push(const void* buf, s64 size) {
+    IN_PUSH_THREAD = true;
+    ON_SCOPE_EXIT(IN_PUSH_THREAD = false);
+
+    while (!m_token.stop_requested()) {
+        mutexLock(&m_mutex);
+        ON_SCOPE_EXIT(mutexUnlock(&m_mutex));
+
+        if (!m_active) {
+            break;
+        }
+
+        if (m_buffer.size() + size >= MAX_BUFFER_SIZE) {
+            svcSleepThread(SLEEPNS);
+        } else {
+            const auto offset = m_buffer.size();
+            m_buffer.resize(offset + size);
+            std::memcpy(m_buffer.data() + offset, buf, size);
+            return true;
+        }
+    }
+
+    return false;
+}
+
+void StreamFtp::Disable() {
+    mutexLock(&m_mutex);
+    ON_SCOPE_EXIT(mutexUnlock(&m_mutex));
+    m_active = false;
+}
+
+Menu::Menu() : MenuBase{"FTP Install (EXPERIMENTAL)"_i18n} {
+    SetAction(Button::B, Action{"Back"_i18n, [this](){
+        SetPop();
+    }});
+
+    mutexInit(&m_mutex);
+    ftpsrv::InitInstallMode(this, OnInstallStart, OnInstallWrite, OnInstallClose);
+}
+
+Menu::~Menu() {
+    // signal for thread to exit and wait.
+    ftpsrv::DisableInstallMode();
+    m_stop_source.request_stop();
+
+    if (m_source) {
+        m_source->Disable();
+    }
+
+    log_write("closing data!!!!\n");
+}
+
+void Menu::Update(Controller* controller, TouchInfo* touch) {
+    MenuBase::Update(controller, touch);
+
+    mutexLock(&m_mutex);
+    ON_SCOPE_EXIT(mutexUnlock(&m_mutex));
+
+    switch (m_state) {
+        case State::None:
+            break;
+
+        case State::Connected:
+            log_write("set to progress\n");
+            m_state = State::Progress;
+            log_write("got connection\n");
+            App::Push(std::make_shared<ui::ProgressBox>("Installing App"_i18n, [this](auto pbox) mutable -> bool {
+                log_write("inside progress box\n");
+                const auto rc = yati::InstallFromSource(pbox, m_source, m_source->m_path);
+                if (R_FAILED(rc)) {
+                    m_source->Disable();
+                    return false;
+                }
+
+                log_write("progress box is done\n");
+                return true;
+            }, [this](bool result){
+                mutexLock(&m_mutex);
+                ON_SCOPE_EXIT(mutexUnlock(&m_mutex));
+
+                if (result) {
+                    App::Notify("Ftp install success!"_i18n);
+                    m_state = State::Done;
+                } else {
+                    App::Notify("Ftp install failed!"_i18n);
+                    m_state = State::Failed;
+                }
+            }));
+            break;
+
+        case State::Progress:
+        case State::Done:
+        case State::Failed:
+            break;
+    }
+}
+
+void Menu::Draw(NVGcontext* vg, Theme* theme) {
+    MenuBase::Draw(vg, theme);
+
+    mutexLock(&m_mutex);
+    ON_SCOPE_EXIT(mutexUnlock(&m_mutex));
+
+    switch (m_state) {
+        case State::None:
+            gfx::drawTextArgs(vg, SCREEN_WIDTH / 2.f, SCREEN_HEIGHT / 2.f, 36.f, NVG_ALIGN_CENTER | NVG_ALIGN_MIDDLE, theme->GetColour(ThemeEntryID_TEXT_INFO), "Waiting for connection..."_i18n.c_str());
+            break;
+
+        case State::Connected:
+            break;
+
+        case State::Progress:
+            gfx::drawTextArgs(vg, SCREEN_WIDTH / 2.f, SCREEN_HEIGHT / 2.f, 36.f, NVG_ALIGN_CENTER | NVG_ALIGN_MIDDLE, theme->GetColour(ThemeEntryID_TEXT_INFO), "Transferring data..."_i18n.c_str());
+            break;
+
+        case State::Done:
+            gfx::drawTextArgs(vg, SCREEN_WIDTH / 2.f, SCREEN_HEIGHT / 2.f, 36.f, NVG_ALIGN_CENTER | NVG_ALIGN_MIDDLE, theme->GetColour(ThemeEntryID_TEXT_INFO), "Press B to exit..."_i18n.c_str());
+            break;
+
+        case State::Failed:
+            gfx::drawTextArgs(vg, SCREEN_WIDTH / 2.f, SCREEN_HEIGHT / 2.f, 36.f, NVG_ALIGN_CENTER | NVG_ALIGN_MIDDLE, theme->GetColour(ThemeEntryID_TEXT_INFO), "Failed to install via FTP, press B to exit..."_i18n.c_str());
+            break;
+    }
+}
+
+void Menu::OnFocusGained() {
+    MenuBase::OnFocusGained();
+}
+
+} // namespace sphaira::ui::menu::ftp

sphaira/source/ui/menus/main_menu.cpp

@@ -3,6 +3,7 @@
 #include "ui/menus/themezer.hpp"
 #include "ui/menus/ghdl.hpp"
 #include "ui/menus/usb_menu.hpp"
+#include "ui/menus/ftp_menu.hpp"
 #include "ui/menus/gc_menu.hpp"
 
 #include "ui/sidebar.hpp"
@@ -321,6 +322,12 @@ MainMenu::MainMenu() {
                 }
 
                 if (App::GetApp()->m_install.Get()) {
+                    if (App::GetFtpEnable()) {
+                        options->Add(std::make_shared<SidebarEntryCallback>("Ftp Install"_i18n, [](){
+                            App::Push(std::make_shared<menu::ftp::Menu>());
+                        }));
+                    }
+
                     options->Add(std::make_shared<SidebarEntryCallback>("Usb Install"_i18n, [](){
                         App::Push(std::make_shared<menu::usb::Menu>());
                     }));

sphaira/source/yati/container/nsp.cpp

@@ -24,14 +24,6 @@ struct Pfs0FileTableEntry {
 
 } // namespace
 
-Result Nsp::Validate(source::Base* source) {
-    u32 magic;
-    u64 bytes_read;
-    R_TRY(source->Read(std::addressof(magic), 0, sizeof(magic), std::addressof(bytes_read)));
-    R_UNLESS(magic == PFS0_MAGIC, 0x1);
-    R_SUCCEED();
-}
-
 Result Nsp::GetCollections(Collections& out) {
     u64 bytes_read;
     s64 off = 0;

sphaira/source/yati/container/xci.cpp

@@ -60,14 +60,6 @@ Result Hfs0GetPartition(source::Base* source, s64 off, Hfs0& out) {
 
 } // namespace
 
-Result Xci::Validate(source::Base* source) {
-    u32 magic;
-    u64 bytes_read;
-    R_TRY(source->Read(std::addressof(magic), 0x100, sizeof(magic), std::addressof(bytes_read)));
-    R_UNLESS(magic == XCI_MAGIC, 0x1);
-    R_SUCCEED();
-}
-
 Result Xci::GetCollections(Collections& out) {
     Hfs0 root{};
     R_TRY(Hfs0GetPartition(m_source.get(), HFS0_HEADER_OFFSET, root));

sphaira/source/yati/source/stream.cpp

@@ -0,0 +1,41 @@
+#include "yati/source/stream.hpp"
+#include "defines.hpp"
+#include "log.hpp"
+
+namespace sphaira::yati::source {
+
+Result Stream::Read(void* _buf, s64 off, s64 size, u64* bytes_read_out) {
+    // streams don't allow for random access (seeking backwards).
+    R_UNLESS(off >= m_offset, 0x1);
+
+    auto buf = static_cast<u8*>(_buf);
+    *bytes_read_out = 0;
+
+    // check if we already have some data in the buffer.
+    while (size) {
+        // while it is invalid to seek backwards, it is valid to seek forwards.
+        // this can be done to skip padding, skip undeeded files etc.
+        // to handle this, simply read the data into a buffer and discard it.
+        if (off > m_offset) {
+            const auto skip_size = off - m_offset;
+            std::vector<u8> temp_buf(skip_size);
+            u64 bytes_read;
+            R_TRY(ReadChunk(temp_buf.data(), temp_buf.size(), &bytes_read));
+
+            m_offset += bytes_read;
+        } else {
+            u64 bytes_read;
+            R_TRY(ReadChunk(buf, size, &bytes_read));
+
+            *bytes_read_out += bytes_read;
+            buf += bytes_read;
+            off += bytes_read;
+            m_offset += bytes_read;
+            size -= bytes_read;
+        }
+    }
+
+    R_SUCCEED();
+}
+
+} // namespace sphaira::yati::source

sphaira/source/yati/source/stream_file.cpp

@@ -0,0 +1,23 @@
+#include "yati/source/stream_file.hpp"
+#include "log.hpp"
+
+namespace sphaira::yati::source {
+
+StreamFile::StreamFile(FsFileSystem* fs, const fs::FsPath& path) {
+    m_open_result = fsFsOpenFile(fs, path, FsOpenMode_Read, std::addressof(m_file));
+}
+
+StreamFile::~StreamFile() {
+    if (R_SUCCEEDED(GetOpenResult())) {
+        fsFileClose(std::addressof(m_file));
+    }
+}
+
+Result StreamFile::ReadChunk(void* buf, s64 size, u64* bytes_read) {
+    R_TRY(GetOpenResult());
+    const auto rc = fsFileRead(std::addressof(m_file), m_offset, buf, size, 0, bytes_read);
+    m_offset += *bytes_read;
+    return rc;
+}
+
+} // namespace sphaira::yati::source

sphaira/source/yati/yati.cpp

@@ -1,5 +1,6 @@
 #include "yati/yati.hpp"
 #include "yati/source/file.hpp"
+#include "yati/source/stream_file.hpp"
 #include "yati/source/stdio.hpp"
 #include "yati/container/nsp.hpp"
 #include "yati/container/xci.hpp"
@@ -36,13 +37,13 @@ struct CustomVectorAllocator {
 public:
     // https://en.cppreference.com/w/cpp/memory/new/operator_new
     auto allocate(std::size_t n) -> T* {
-        log_write("allocating ptr size: %zu\n", n);
+        // log_write("allocating ptr size: %zu\n", n);
         return new(align) T[n];
     }
 
     // https://en.cppreference.com/w/cpp/memory/new/operator_delete
     auto deallocate(T* p, std::size_t n) noexcept -> void {
-        log_write("deleting ptr size: %zu\n", n);
+        // log_write("deleting ptr size: %zu\n", n);
         ::operator delete[] (p, n, align);
     }
 
@@ -62,17 +63,6 @@ using PageAlignedVector = std::vector<u8, PageAllocator<u8>>;
 
 constexpr u32 KEYGEN_LIMIT = 0x20;
 
-#if 0
-struct FwVersion {
-    u32 value;
-    auto relstep() const -> u8 { return (value >> 0)  & 0xFFFF; }
-    auto micro() const -> u8 { return (value >> 16) & 0x000F; }
-    auto minor() const -> u8 { return (value >> 20) & 0x003F; }
-    auto major() const -> u8 { return (value >> 26) & 0x003F; }
-    auto hos() const -> u32 { return MAKEHOSVERSION(major(), minor(), micro()); }
-};
-#endif
-
 struct NcaCollection : container::CollectionEntry {
     // NcmContentType
     u8 type{};
@@ -293,6 +283,14 @@ struct Yati {
     Result decompressFuncInternal(ThreadData* t);
     Result writeFuncInternal(ThreadData* t);
 
+    Result ParseTicketsIntoCollection(std::vector<TikCollection>& tickets, const container::Collections& collections, bool read_data);
+    Result GetLatestVersion(const CnmtCollection& cnmt, u32& version_out, bool& skip);
+    Result ShouldSkip(const CnmtCollection& cnmt, bool& skip);
+    Result ImportTickets(std::span<TikCollection> tickets);
+    Result RemoveInstalledNcas(const CnmtCollection& cnmt);
+    Result RegisterNcasAndPushRecord(const CnmtCollection& cnmt, u32 latest_version_num);
+
+
 // private:
     ui::ProgressBox* pbox{};
     std::shared_ptr<source::Base> source{};
@@ -362,8 +360,14 @@ HashStr hexIdToStr(auto id) {
 // read thread reads all data from the source, it also handles
 // parsing ncz headers, sections and reading ncz blocks
 Result Yati::readFuncInternal(ThreadData* t) {
+    // the main buffer which data is read into.
     PageAlignedVector buf;
+    // workaround ncz block reading ahead. if block isn't found, we usually
+    // would seek back to the offset, however this is not possible in stream
+    // mode, so we instead store the data to the temp buffer and pre-pend it.
+    PageAlignedVector temp_buf;
     buf.reserve(t->max_buffer_size);
+    temp_buf.reserve(t->max_buffer_size);
 
     while (t->read_offset < t->nca->size && R_SUCCEEDED(t->GetResults())) {
         const auto buffer_offset = t->read_offset;
@@ -374,10 +378,18 @@ Result Yati::readFuncInternal(ThreadData* t) {
             read_size = NCZ_SECTION_OFFSET;
         }
 
+        s64 buf_offset = 0;
+        if (!temp_buf.empty()) {
+            buf = temp_buf;
+            read_size -= temp_buf.size();
+            buf_offset = temp_buf.size();
+            temp_buf.clear();
+        }
+
         u64 bytes_read{};
-        buf.resize(read_size);
-        R_TRY(t->Read(buf.data(), read_size, std::addressof(bytes_read)));
-        auto buf_size = bytes_read;
+        buf.resize(buf_offset + read_size);
+        R_TRY(t->Read(buf.data() + buf_offset, read_size, std::addressof(bytes_read)));
+        auto buf_size = buf_offset + bytes_read;
 
         // read enough bytes for ncz, check magic
         if (t->read_offset == NCZ_SECTION_OFFSET) {
@@ -394,10 +406,12 @@ Result Yati::readFuncInternal(ThreadData* t) {
                 R_TRY(t->Read(t->ncz_sections.data(), t->ncz_sections.size() * sizeof(ncz::Section), std::addressof(bytes_read)));
 
                 // check for ncz block header.
-                const auto read_off = t->read_offset;
                 R_TRY(t->Read(std::addressof(t->ncz_block_header), sizeof(t->ncz_block_header), std::addressof(bytes_read)));
                 if (t->ncz_block_header.magic != NCZ_BLOCK_MAGIC) {
-                    t->read_offset = read_off;
+                    // didn't find block, keep the data we just read in the temp buffer.
+                    temp_buf.resize(sizeof(t->ncz_block_header));
+                    std::memcpy(temp_buf.data(), std::addressof(t->ncz_block_header), temp_buf.size());
+                    log_write("storing temp data of size: %zu\n", temp_buf.size());
                 } else {
                     // validate block header.
                     R_UNLESS(t->ncz_block_header.version == 0x2, Result_InvalidNczBlockVersion);
@@ -891,7 +905,6 @@ Result Yati::InstallNca(std::span<TikCollection> tickets, NcaCollection& nca) {
     std::memcpy(std::addressof(content_id), nca.hash, sizeof(content_id));
 
     log_write("old id: %s new id: %s\n", hexIdToStr(nca.content_id).str, hexIdToStr(content_id).str);
-    log_write("doing register: %s\n", nca.name.c_str());
     if (!config.skip_nca_hash_verify && !nca.modified) {
         if (std::memcmp(&nca.content_id, nca.hash, sizeof(nca.content_id))) {
             log_write("nca hash is invalid!!!!\n");
@@ -1027,11 +1040,7 @@ Result Yati::InstallControlNca(std::span<TikCollection> tickets, const CnmtColle
     R_SUCCEED();
 }
 
-Result InstallInternal(ui::ProgressBox* pbox, std::shared_ptr<source::Base> source, const container::Collections& collections) {
-    auto yati = std::make_unique<Yati>(pbox, source);
-    R_TRY(yati->Setup());
-
-    std::vector<TikCollection> tickets{};
+Result Yati::ParseTicketsIntoCollection(std::vector<TikCollection>& tickets, const container::Collections& collections, bool read_data) {
     for (const auto& collection : collections) {
         if (collection.name.ends_with(".tik")) {
             TikCollection entry{};
@@ -1046,17 +1055,228 @@ Result InstallInternal(ui::ProgressBox* pbox, std::shared_ptr<source::Base> sour
             entry.ticket.resize(collection.size);
             entry.cert.resize(cert->size);
 
-            u64 bytes_read;
-            R_TRY(source->Read(entry.ticket.data(), collection.offset, entry.ticket.size(), &bytes_read));
-            R_TRY(source->Read(entry.cert.data(), cert->offset, entry.cert.size(), &bytes_read));
+            // only supported on non-stream installs.
+            if (read_data) {
+                u64 bytes_read;
+                R_TRY(source->Read(entry.ticket.data(), collection.offset, entry.ticket.size(), &bytes_read));
+                R_TRY(source->Read(entry.cert.data(), cert->offset, entry.cert.size(), &bytes_read));
+            }
+
             tickets.emplace_back(entry);
         }
     }
 
+    R_SUCCEED();
+}
+
+Result Yati::GetLatestVersion(const CnmtCollection& cnmt, u32& version_out, bool& skip) {
+    const auto app_id = ncm::GetAppId(cnmt.key);
+
+    bool has_records;
+    R_TRY(nsIsAnyApplicationEntityInstalled(app_id, &has_records));
+
+    // TODO: fix this when gamecard is inserted as it will only return records
+    // for the gamecard...
+    // may have to use ncm directly to get the keys, then parse that.
+    u32 latest_version_num = cnmt.key.version;
+    if (has_records) {
+        s32 meta_count{};
+        R_TRY(nsCountApplicationContentMeta(app_id, &meta_count));
+        R_UNLESS(meta_count > 0, 0x1);
+
+        std::vector<ncm::ContentStorageRecord> records(meta_count);
+        s32 count;
+        R_TRY(ns::ListApplicationRecordContentMeta(std::addressof(ns_app), 0, app_id, records.data(), records.size(), &count));
+        R_UNLESS(count == records.size(), 0x1);
+
+        for (auto& record : records) {
+            log_write("found record: 0x%016lX type: %u version: %u\n", record.key.id, record.key.type, record.key.version);
+            log_write("cnmt record: 0x%016lX type: %u version: %u\n", cnmt.key.id, cnmt.key.type, cnmt.key.version);
+
+            if (record.key.id == cnmt.key.id && cnmt.key.version == record.key.version && config.skip_if_already_installed) {
+                log_write("skipping as already installed\n");
+                skip = true;
+            }
+
+            // check if we are downgrading
+            if (cnmt.key.type == NcmContentMetaType_Patch) {
+                if (cnmt.key.type == record.key.type && cnmt.key.version < record.key.version && !config.allow_downgrade) {
+                    log_write("skipping due to it being lower\n");
+                    skip = true;
+                }
+            } else {
+                latest_version_num = std::max(latest_version_num, record.key.version);
+            }
+        }
+    }
+
+    R_SUCCEED();
+}
+
+Result Yati::ShouldSkip(const CnmtCollection& cnmt, bool& skip) {
+    // skip invalid types
+    if (!(cnmt.key.type & 0x80)) {
+        log_write("\tskipping: invalid: %u\n", cnmt.key.type);
+        skip = true;
+    } else if (config.skip_base && cnmt.key.type == NcmContentMetaType_Application) {
+        log_write("\tskipping: [NcmContentMetaType_Application]\n");
+        skip = true;
+    } else if (config.skip_patch && cnmt.key.type == NcmContentMetaType_Patch) {
+        log_write("\tskipping: [NcmContentMetaType_Application]\n");
+        skip = true;
+    } else if (config.skip_addon && cnmt.key.type == NcmContentMetaType_AddOnContent) {
+        log_write("\tskipping: [NcmContentMetaType_AddOnContent]\n");
+        skip = true;
+    } else if (config.skip_data_patch && cnmt.key.type == NcmContentMetaType_DataPatch) {
+        log_write("\tskipping: [NcmContentMetaType_DataPatch]\n");
+        skip = true;
+    }
+
+    R_SUCCEED();
+}
+
+Result Yati::ImportTickets(std::span<TikCollection> tickets) {
+    for (auto& ticket : tickets) {
+        if (ticket.required) {
+            if (config.skip_ticket) {
+                log_write("WARNING: skipping ticket install, but it's required!\n");
+            } else {
+                log_write("patching ticket\n");
+                if (config.patch_ticket) {
+                    R_TRY(es::PatchTicket(ticket.ticket, keys, false));
+                }
+                log_write("installing ticket\n");
+                R_TRY(es::ImportTicket(std::addressof(es), ticket.ticket.data(), ticket.ticket.size(), ticket.cert.data(), ticket.cert.size()));
+                ticket.required = false;
+            }
+        }
+    }
+
+    R_SUCCEED();
+}
+
+Result Yati::RemoveInstalledNcas(const CnmtCollection& cnmt) {
+    const auto app_id = ncm::GetAppId(cnmt.key);
+
+    // remove current entries (if any).
+    s32 db_list_total;
+    s32 db_list_count;
+    u64 id_min = cnmt.key.id;
+    u64 id_max = cnmt.key.id;
+    std::vector<NcmContentMetaKey> keys(1);
+
+    // if installing a patch, remove all previously installed patches.
+    if (cnmt.key.type == NcmContentMetaType_Patch) {
+        id_min = 0;
+        id_max = UINT64_MAX;
+    }
+
+    log_write("listing keys\n");
+    for (size_t i = 0; i < std::size(NCM_STORAGE_IDS); i++) {
+        auto& cs = ncm_cs[i];
+        auto& db = ncm_db[i];
+
+        std::vector<NcmContentMetaKey> keys(1);
+        R_TRY(ncmContentMetaDatabaseList(std::addressof(db), std::addressof(db_list_total), std::addressof(db_list_count), keys.data(), keys.size(), static_cast<NcmContentMetaType>(cnmt.key.type), app_id, id_min, id_max, NcmContentInstallType_Full));
+
+        if (db_list_total != keys.size()) {
+            keys.resize(db_list_total);
+            if (keys.size()) {
+                R_TRY(ncmContentMetaDatabaseList(std::addressof(db), std::addressof(db_list_total), std::addressof(db_list_count), keys.data(), keys.size(), static_cast<NcmContentMetaType>(cnmt.key.type), app_id, id_min, id_max, NcmContentInstallType_Full));
+            }
+        }
+
+        for (auto& key : keys) {
+            log_write("found key: 0x%016lX type: %u version: %u\n", key.id, key.type, key.version);
+            NcmContentMetaHeader header;
+            u64 out_size;
+            log_write("trying to get from db\n");
+            R_TRY(ncmContentMetaDatabaseGet(std::addressof(db), std::addressof(key), std::addressof(out_size), std::addressof(header), sizeof(header)));
+            R_UNLESS(out_size == sizeof(header), Result_NcmDbCorruptHeader);
+            log_write("trying to list infos\n");
+
+            std::vector<NcmContentInfo> infos(header.content_count);
+            s32 content_info_out;
+            R_TRY(ncmContentMetaDatabaseListContentInfo(std::addressof(db), std::addressof(content_info_out), infos.data(), infos.size(), std::addressof(key), 0));
+            R_UNLESS(content_info_out == infos.size(), Result_NcmDbCorruptInfos);
+            log_write("size matches\n");
+
+            for (auto& info : infos) {
+                R_TRY(ncm::Delete(std::addressof(cs), std::addressof(info.content_id)));
+            }
+
+            log_write("trying to remove it\n");
+            R_TRY(ncmContentMetaDatabaseRemove(std::addressof(db), std::addressof(key)));
+            R_TRY(ncmContentMetaDatabaseCommit(std::addressof(db)));
+            log_write("all done with this key\n\n");
+        }
+    }
+
+    log_write("done with keys\n");
+    R_SUCCEED();
+}
+
+Result Yati::RegisterNcasAndPushRecord(const CnmtCollection& cnmt, u32 latest_version_num) {
+    const auto app_id = ncm::GetAppId(cnmt.key);
+
+    // register all nca's
+    log_write("registering cnmt nca\n");
+    R_TRY(ncm::Register(std::addressof(cs), std::addressof(cnmt.content_id), std::addressof(cnmt.placeholder_id)));
+    log_write("registered cnmt nca\n");
+
+    for (auto& nca : cnmt.ncas) {
+        if (nca.type != NcmContentType_DeltaFragment) {
+            log_write("registering nca: %s\n", nca.name.c_str());
+            R_TRY(ncm::Register(std::addressof(cs), std::addressof(nca.content_id), std::addressof(nca.placeholder_id)));
+            log_write("registered nca: %s\n", nca.name.c_str());
+        }
+    }
+
+    log_write("register'd all ncas\n");
+
+    // build ncm meta and push to the database.
+    BufHelper buf{};
+    buf.write(std::addressof(cnmt.header), sizeof(cnmt.header));
+    buf.write(cnmt.extended_header.data(), cnmt.extended_header.size());
+    buf.write(std::addressof(cnmt.content_info), sizeof(cnmt.content_info));
+
+    for (auto& info : cnmt.infos) {
+        buf.write(std::addressof(info.info), sizeof(info.info));
+    }
+
+    pbox->NewTransfer("Updating ncm databse"_i18n);
+    R_TRY(ncmContentMetaDatabaseSet(std::addressof(db), std::addressof(cnmt.key), buf.buf.data(), buf.tell()));
+    R_TRY(ncmContentMetaDatabaseCommit(std::addressof(db)));
+
+    // push record.
+    ncm::ContentStorageRecord content_storage_record{};
+    content_storage_record.key = cnmt.key;
+    content_storage_record.storage_id = storage_id;
+    pbox->NewTransfer("Pushing application record"_i18n);
+
+    R_TRY(ns::PushApplicationRecord(std::addressof(ns_app), app_id, std::addressof(content_storage_record), 1));
+    if (hosversionAtLeast(6,0,0)) {
+        R_TRY(avmInitialize());
+        ON_SCOPE_EXIT(avmExit());
+
+        R_TRY(avmPushLaunchVersion(app_id, latest_version_num));
+    }
+    log_write("pushed\n");
+
+    R_SUCCEED();
+}
+
+Result InstallInternal(ui::ProgressBox* pbox, std::shared_ptr<source::Base> source, const container::Collections& collections) {
+    auto yati = std::make_unique<Yati>(pbox, source);
+    R_TRY(yati->Setup());
+
+    std::vector<TikCollection> tickets{};
+    R_TRY(yati->ParseTicketsIntoCollection(tickets, collections, true));
+
     std::vector<CnmtCollection> cnmts{};
     for (const auto& collection : collections) {
         log_write("found collection: %s\n", collection.name.c_str());
-        if (collection.name.ends_with(".cnmt.nca")) {
+        if (collection.name.ends_with(".cnmt.nca") || collection.name.ends_with(".cnmt.ncz")) {
             auto& cnmt = cnmts.emplace_back(NcaCollection{collection});
             cnmt.type = NcmContentType_Meta;
         }
@@ -1072,63 +1292,10 @@ Result InstallInternal(ui::ProgressBox* pbox, std::shared_ptr<source::Base> sour
 
         R_TRY(yati->InstallCnmtNca(tickets, cnmt, collections));
 
+        u32 latest_version_num;
         bool skip = false;
-        const auto app_id = ncm::GetAppId(cnmt.key);
-        bool has_records;
-        R_TRY(nsIsAnyApplicationEntityInstalled(app_id, &has_records));
-
-        // TODO: fix this when gamecard is inserted as it will only return records
-        // for the gamecard...
-        // may have to use ncm directly to get the keys, then parse that.
-        u32 latest_version_num = cnmt.key.version;
-        if (has_records) {
-            s32 meta_count{};
-            R_TRY(nsCountApplicationContentMeta(app_id, &meta_count));
-            R_UNLESS(meta_count > 0, 0x1);
-
-            std::vector<ncm::ContentStorageRecord> records(meta_count);
-            s32 count;
-            R_TRY(ns::ListApplicationRecordContentMeta(std::addressof(yati->ns_app), 0, app_id, records.data(), records.size(), &count));
-            R_UNLESS(count == records.size(), 0x1);
-
-            for (auto& record : records) {
-                log_write("found record: 0x%016lX type: %u version: %u\n", record.key.id, record.key.type, record.key.version);
-                log_write("cnmt record: 0x%016lX type: %u version: %u\n", cnmt.key.id, cnmt.key.type, cnmt.key.version);
-
-                if (record.key.id == cnmt.key.id && cnmt.key.version == record.key.version && yati->config.skip_if_already_installed) {
-                    log_write("skipping as already installed\n");
-                    skip = true;
-                }
-
-                // check if we are downgrading
-                if (cnmt.key.type == NcmContentMetaType_Patch) {
-                    if (cnmt.key.type == record.key.type && cnmt.key.version < record.key.version && !yati->config.allow_downgrade) {
-                        log_write("skipping due to it being lower\n");
-                        skip = true;
-                    }
-                } else {
-                    latest_version_num = std::max(latest_version_num, record.key.version);
-                }
-            }
-        }
-
-        // skip invalid types
-        if (!(cnmt.key.type & 0x80)) {
-            log_write("\tskipping: invalid: %u\n", cnmt.key.type);
-            skip = true;
-        } else if (yati->config.skip_base && cnmt.key.type == NcmContentMetaType_Application) {
-            log_write("\tskipping: [NcmContentMetaType_Application]\n");
-            skip = true;
-        } else if (yati->config.skip_patch && cnmt.key.type == NcmContentMetaType_Patch) {
-            log_write("\tskipping: [NcmContentMetaType_Application]\n");
-            skip = true;
-        } else if (yati->config.skip_addon && cnmt.key.type == NcmContentMetaType_AddOnContent) {
-            log_write("\tskipping: [NcmContentMetaType_AddOnContent]\n");
-            skip = true;
-        } else if (yati->config.skip_data_patch && cnmt.key.type == NcmContentMetaType_DataPatch) {
-            log_write("\tskipping: [NcmContentMetaType_DataPatch]\n");
-            skip = true;
-        }
+        R_TRY(yati->GetLatestVersion(cnmt, latest_version_num, skip));
+        R_TRY(yati->ShouldSkip(cnmt, skip));
 
         if (skip) {
             log_write("skipping install!\n");
@@ -1145,125 +1312,104 @@ Result InstallInternal(ui::ProgressBox* pbox, std::shared_ptr<source::Base> sour
             }
         }
 
-        // log_write("exiting early :)\n");
-        // return 0;
+        R_TRY(yati->ImportTickets(tickets));
+        R_TRY(yati->RemoveInstalledNcas(cnmt));
+        R_TRY(yati->RegisterNcasAndPushRecord(cnmt, latest_version_num));
+    }
 
-        for (auto& ticket : tickets) {
-            if (ticket.required) {
-                if (yati->config.skip_ticket) {
-                    log_write("WARNING: skipping ticket install, but it's required!\n");
-                } else {
-                    log_write("patching ticket\n");
-                    if (yati->config.patch_ticket) {
-                        R_TRY(es::PatchTicket(ticket.ticket, yati->keys, false));
-                    }
-                    log_write("installing ticket\n");
-                    R_TRY(es::ImportTicket(std::addressof(yati->es), ticket.ticket.data(), ticket.ticket.size(), ticket.cert.data(), ticket.cert.size()));
-                    ticket.required = false;
-                }
+    log_write("success!\n");
+    R_SUCCEED();
+}
+
+Result InstallInternalStream(ui::ProgressBox* pbox, std::shared_ptr<source::Base> source, container::Collections collections) {
+    auto yati = std::make_unique<Yati>(pbox, source);
+    R_TRY(yati->Setup());
+
+    // not supported with stream installs (yet).
+    yati->config.convert_to_standard_crypto = false;
+    yati->config.lower_master_key = false;
+
+    std::vector<NcaCollection> ncas{};
+    std::vector<CnmtCollection> cnmts{};
+    std::vector<TikCollection> tickets{};
+
+    ON_SCOPE_EXIT(
+        for (const auto& cnmt : cnmts) {
+            ncmContentStorageDeletePlaceHolder(std::addressof(yati->cs), std::addressof(cnmt.placeholder_id));
+        }
+
+        for (const auto& nca : ncas) {
+            ncmContentStorageDeletePlaceHolder(std::addressof(yati->cs), std::addressof(nca.placeholder_id));
+        }
+    );
+
+    // fill ticket entries, the data will be filled later on.
+    R_TRY(yati->ParseTicketsIntoCollection(tickets, collections, false));
+
+    // sort based on lowest offset.
+    const auto sorter = [](const container::CollectionEntry& lhs, const container::CollectionEntry& rhs) -> bool {
+        return lhs.offset < rhs.offset;
+    };
+
+    std::sort(collections.begin(), collections.end(), sorter);
+
+    for (const auto& collection : collections) {
+        if (collection.name.ends_with(".nca") || collection.name.ends_with(".ncz")) {
+            auto& nca = ncas.emplace_back(NcaCollection{collection});
+            if (collection.name.ends_with(".cnmt.nca") || collection.name.ends_with(".cnmt.ncz")) {
+                auto& cnmt = cnmts.emplace_back(nca);
+                cnmt.type = NcmContentType_Meta;
+                R_TRY(yati->InstallCnmtNca(tickets, cnmt, collections));
+            } else {
+                R_TRY(yati->InstallNca(tickets, nca));
+            }
+        } else if (collection.name.ends_with(".tik") || collection.name.ends_with(".cert")) {
+            FsRightsId rights_id{};
+            keys::parse_hex_key(rights_id.c, collection.name.c_str());
+            const auto str = collection.name.substr(0, collection.name.length() - 4) + ".cert";
+
+            auto entry = std::find_if(tickets.begin(), tickets.end(), [rights_id](auto& e){
+                return !std::memcmp(&rights_id, &e.rights_id, sizeof(rights_id));
+            });
+
+            // this will never fail...but just in case.
+            R_UNLESS(entry != tickets.end(), Result_CertNotFound);
+
+            u64 bytes_read;
+            if (collection.name.ends_with(".tik")) {
+                R_TRY(source->Read(entry->ticket.data(), collection.offset, entry->ticket.size(), &bytes_read));
+            } else {
+                R_TRY(source->Read(entry->cert.data(), collection.offset, entry->cert.size(), &bytes_read));
             }
         }
+    }
 
-        log_write("listing keys\n");
+    for (auto& cnmt : cnmts) {
+        // copy nca structs into cnmt.
+        for (auto& cnmt_nca : cnmt.ncas) {
+            auto it = std::find_if(ncas.cbegin(), ncas.cend(), [cnmt_nca](auto& e){
+                return e.name == cnmt_nca.name;
+            });
 
-        // remove current entries (if any).
-        s32 db_list_total;
-        s32 db_list_count;
-        u64 id_min = cnmt.key.id;
-        u64 id_max = cnmt.key.id;
-        std::vector<NcmContentMetaKey> keys(1);
-
-        // if installing a patch, remove all previously installed patches.
-        if (cnmt.key.type == NcmContentMetaType_Patch) {
-            id_min = 0;
-            id_max = UINT64_MAX;
+            R_UNLESS(it != ncas.cend(), Result_NczSectionNotFound);
+            const auto type = cnmt_nca.type;
+            cnmt_nca = *it;
+            cnmt_nca.type = type;
         }
 
-        for (size_t i = 0; i < std::size(NCM_STORAGE_IDS); i++) {
-            auto& cs = yati->ncm_cs[i];
-            auto& db = yati->ncm_db[i];
+        u32 latest_version_num;
+        bool skip = false;
+        R_TRY(yati->GetLatestVersion(cnmt, latest_version_num, skip));
+        R_TRY(yati->ShouldSkip(cnmt, skip));
 
-            std::vector<NcmContentMetaKey> keys(1);
-            R_TRY(ncmContentMetaDatabaseList(std::addressof(db), std::addressof(db_list_total), std::addressof(db_list_count), keys.data(), keys.size(), static_cast<NcmContentMetaType>(cnmt.key.type), app_id, id_min, id_max, NcmContentInstallType_Full));
-
-            if (db_list_total != keys.size()) {
-                keys.resize(db_list_total);
-                if (keys.size()) {
-                    R_TRY(ncmContentMetaDatabaseList(std::addressof(db), std::addressof(db_list_total), std::addressof(db_list_count), keys.data(), keys.size(), static_cast<NcmContentMetaType>(cnmt.key.type), app_id, id_min, id_max, NcmContentInstallType_Full));
-                }
-            }
-
-            for (auto& key : keys) {
-                log_write("found key: 0x%016lX type: %u version: %u\n", key.id, key.type, key.version);
-                NcmContentMetaHeader header;
-                u64 out_size;
-                log_write("trying to get from db\n");
-                R_TRY(ncmContentMetaDatabaseGet(std::addressof(db), std::addressof(key), std::addressof(out_size), std::addressof(header), sizeof(header)));
-                R_UNLESS(out_size == sizeof(header), Result_NcmDbCorruptHeader);
-                log_write("trying to list infos\n");
-
-                std::vector<NcmContentInfo> infos(header.content_count);
-                s32 content_info_out;
-                R_TRY(ncmContentMetaDatabaseListContentInfo(std::addressof(db), std::addressof(content_info_out), infos.data(), infos.size(), std::addressof(key), 0));
-                R_UNLESS(content_info_out == infos.size(), Result_NcmDbCorruptInfos);
-                log_write("size matches\n");
-
-                for (auto& info : infos) {
-                    R_TRY(ncm::Delete(std::addressof(cs), std::addressof(info.content_id)));
-                }
-
-                log_write("trying to remove it\n");
-                R_TRY(ncmContentMetaDatabaseRemove(std::addressof(db), std::addressof(key)));
-                R_TRY(ncmContentMetaDatabaseCommit(std::addressof(db)));
-                log_write("all done with this key\n\n");
-            }
+        if (skip) {
+            log_write("skipping install!\n");
+            continue;
         }
 
-        log_write("done with keys\n");
-
-        // register all nca's
-        log_write("registering cnmt nca\n");
-        R_TRY(ncm::Register(std::addressof(yati->cs), std::addressof(cnmt.content_id), std::addressof(cnmt.placeholder_id)));
-        log_write("registered cnmt nca\n");
-
-        for (auto& nca : cnmt.ncas) {
-            log_write("registering nca: %s\n", nca.name.c_str());
-            R_TRY(ncm::Register(std::addressof(yati->cs), std::addressof(nca.content_id), std::addressof(nca.placeholder_id)));
-            log_write("registered nca: %s\n", nca.name.c_str());
-        }
-
-        log_write("register'd all ncas\n");
-
-        {
-            BufHelper buf{};
-            buf.write(std::addressof(cnmt.header), sizeof(cnmt.header));
-            buf.write(cnmt.extended_header.data(), cnmt.extended_header.size());
-            buf.write(std::addressof(cnmt.content_info), sizeof(cnmt.content_info));
-
-            for (auto& info : cnmt.infos) {
-                buf.write(std::addressof(info.info), sizeof(info.info));
-            }
-
-            pbox->NewTransfer("Updating ncm databse"_i18n);
-            R_TRY(ncmContentMetaDatabaseSet(std::addressof(yati->db), std::addressof(cnmt.key), buf.buf.data(), buf.tell()));
-            R_TRY(ncmContentMetaDatabaseCommit(std::addressof(yati->db)));
-        }
-
-        {
-            ncm::ContentStorageRecord content_storage_record{};
-            content_storage_record.key = cnmt.key;
-            content_storage_record.storage_id = yati->storage_id;
-            pbox->NewTransfer("Pushing application record"_i18n);
-
-            R_TRY(ns::PushApplicationRecord(std::addressof(yati->ns_app), app_id, std::addressof(content_storage_record), 1));
-            if (hosversionAtLeast(6,0,0)) {
-                R_TRY(avmInitialize());
-                ON_SCOPE_EXIT(avmExit());
-
-                R_TRY(avmPushLaunchVersion(app_id, latest_version_num));
-            }
-            log_write("pushed\n");
-        }
+        R_TRY(yati->ImportTickets(tickets));
+        R_TRY(yati->RemoveInstalledNcas(cnmt));
+        R_TRY(yati->RegisterNcasAndPushRecord(cnmt, latest_version_num));
     }
 
     log_write("success!\n");
@@ -1274,6 +1420,7 @@ Result InstallInternal(ui::ProgressBox* pbox, std::shared_ptr<source::Base> sour
 
 Result InstallFromFile(ui::ProgressBox* pbox, FsFileSystem* fs, const fs::FsPath& path) {
     return InstallFromSource(pbox, std::make_shared<source::File>(fs, path), path);
+    // return InstallFromSource(pbox, std::make_shared<source::StreamFile>(fs, path), path);
 }
 
 Result InstallFromStdioFile(ui::ProgressBox* pbox, const fs::FsPath& path) {
@@ -1281,16 +1428,16 @@ Result InstallFromStdioFile(ui::ProgressBox* pbox, const fs::FsPath& path) {
 }
 
 Result InstallFromSource(ui::ProgressBox* pbox, std::shared_ptr<source::Base> source, const fs::FsPath& path) {
-    if (R_SUCCEEDED(container::Nsp::Validate(source.get()))) {
-        log_write("found nsp\n");
+    const auto ext = std::strrchr(path.s, '.');
+    R_UNLESS(ext, Result_ContainerNotFound);
+
+    if (!strcasecmp(ext, ".nsp") || !strcasecmp(ext, ".nsz")) {
         return InstallFromContainer(pbox, std::make_unique<container::Nsp>(source));
-    } else if (R_SUCCEEDED(container::Xci::Validate(source.get()))) {
-        log_write("found xci\n");
+    } else if (!strcasecmp(ext, ".xci") || !strcasecmp(ext, ".xcz")) {
         return InstallFromContainer(pbox, std::make_unique<container::Xci>(source));
-    } else {
-        log_write("found unknown container\n");
-        R_THROW(Result_ContainerNotFound);
     }
+
+    R_THROW(Result_ContainerNotFound);
 }
 
 Result InstallFromContainer(ui::ProgressBox* pbox, std::shared_ptr<container::Base> container) {
@@ -1300,7 +1447,11 @@ Result InstallFromContainer(ui::ProgressBox* pbox, std::shared_ptr<container::Ba
 }
 
 Result InstallFromCollections(ui::ProgressBox* pbox, std::shared_ptr<source::Base> source, const container::Collections& collections) {
-    return InstallInternal(pbox, source, collections);
+    if (source->IsStream()) {
+        return InstallInternalStream(pbox, source, collections);
+    } else {
+        return InstallInternal(pbox, source, collections);
+    }
 }
 
 } // namespace sphaira::yati