diff --git a/configure.ac b/configure.ac
index 94da5a76a59..44f7dda0ab0 100644
--- a/configure.ac
+++ b/configure.ac
@@ -164,6 +164,7 @@ AC_CONFIG_FILES([Makefile
                  pyanaconda/modules/payloads/Makefile
                  pyanaconda/modules/payloads/payload/Makefile
                  pyanaconda/modules/payloads/payload/dnf/Makefile
+                 pyanaconda/modules/payloads/payload/flatpak/Makefile
                  pyanaconda/modules/payloads/payload/live_os/Makefile
                  pyanaconda/modules/payloads/payload/live_image/Makefile
                  pyanaconda/modules/payloads/payload/rpm_ostree/Makefile
diff --git a/pyanaconda/core/constants.py b/pyanaconda/core/constants.py
index 0f333273b76..645b3f9990e 100644
--- a/pyanaconda/core/constants.py
+++ b/pyanaconda/core/constants.py
@@ -395,6 +395,7 @@ class DisplayModes(Enum):
 
 # Types of the payload.
 PAYLOAD_TYPE_DNF = "DNF"
+PAYLOAD_TYPE_FLATPAK = "FLATPAK"
 PAYLOAD_TYPE_LIVE_OS = "LIVE_OS"
 PAYLOAD_TYPE_LIVE_IMAGE = "LIVE_IMAGE"
 PAYLOAD_TYPE_RPM_OSTREE = "RPM_OSTREE"
diff --git a/pyanaconda/modules/common/constants/interfaces.py b/pyanaconda/modules/common/constants/interfaces.py
index a143b0459c4..852255a0e8c 100644
--- a/pyanaconda/modules/common/constants/interfaces.py
+++ b/pyanaconda/modules/common/constants/interfaces.py
@@ -69,6 +69,11 @@
     basename="DNF"
 )
 
+PAYLOAD_FLATPAK = DBusInterfaceIdentifier(
+    namespace=PAYLOAD_NAMESPACE,
+    basename="FLATPAK"
+)
+
 PAYLOAD_LIVE_IMAGE = DBusInterfaceIdentifier(
     namespace=PAYLOAD_NAMESPACE,
     basename="LiveImage"
diff --git a/pyanaconda/modules/payloads/constants.py b/pyanaconda/modules/payloads/constants.py
index afa6b27d5fb..94fabad5cb0 100644
--- a/pyanaconda/modules/payloads/constants.py
+++ b/pyanaconda/modules/payloads/constants.py
@@ -19,7 +19,7 @@
 from enum import Enum, unique, auto
 
 from pyanaconda.core.constants import \
-    PAYLOAD_TYPE_DNF, PAYLOAD_TYPE_LIVE_OS, PAYLOAD_TYPE_LIVE_IMAGE, \
+    PAYLOAD_TYPE_DNF, PAYLOAD_TYPE_FLATPAK, PAYLOAD_TYPE_LIVE_OS, PAYLOAD_TYPE_LIVE_IMAGE, \
     SOURCE_TYPE_LIVE_OS_IMAGE, SOURCE_TYPE_HMC, SOURCE_TYPE_CDROM, SOURCE_TYPE_REPO_FILES, \
     SOURCE_TYPE_REPO_PATH, SOURCE_TYPE_NFS, SOURCE_TYPE_URL, SOURCE_TYPE_HDD, SOURCE_TYPE_CDN, \
     SOURCE_TYPE_CLOSEST_MIRROR, PAYLOAD_TYPE_RPM_OSTREE, \
@@ -37,6 +37,7 @@
 class PayloadType(Enum):
     """Type of the payload."""
     DNF = PAYLOAD_TYPE_DNF
+    FLATPAK = PAYLOAD_TYPE_FLATPAK
     LIVE_OS = PAYLOAD_TYPE_LIVE_OS
     LIVE_IMAGE = PAYLOAD_TYPE_LIVE_IMAGE
     RPM_OSTREE = PAYLOAD_TYPE_RPM_OSTREE
diff --git a/pyanaconda/modules/payloads/payload/Makefile.am b/pyanaconda/modules/payloads/payload/Makefile.am
index 5bc389bfacf..26906157cda 100644
--- a/pyanaconda/modules/payloads/payload/Makefile.am
+++ b/pyanaconda/modules/payloads/payload/Makefile.am
@@ -14,7 +14,7 @@
 # You should have received a copy of the GNU Lesser General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-SUBDIRS = dnf live_os live_image rpm_ostree
+SUBDIRS = dnf flatpak live_os live_image rpm_ostree
 
 pkgpyexecdir = $(pyexecdir)/py$(PACKAGE_NAME)
 dnf_moduledir = $(pkgpyexecdir)/modules/payloads/payload
diff --git a/pyanaconda/modules/payloads/payload/dnf/dnf.py b/pyanaconda/modules/payloads/payload/dnf/dnf.py
index 8e15d5c0451..22fa5519db4 100644
--- a/pyanaconda/modules/payloads/payload/dnf/dnf.py
+++ b/pyanaconda/modules/payloads/payload/dnf/dnf.py
@@ -371,6 +371,16 @@ def calculate_required_space(self):
                                                   self._dnf_manager.get_installation_size())
         return required_space.get_bytes()
 
+    def needs_flatpak_side_payload(self):
+        return True
+
+    def get_flatpak_refs(self):
+        """Get the list of Flatpak refs to install.
+
+        :return: list of Flatpak refs
+        """
+        return self._dnf_manager.get_flatpak_refs()
+
     def get_repo_configurations(self):
         """Get RepoConfiguration structures for all sources.
 
diff --git a/pyanaconda/modules/payloads/payload/dnf/dnf_manager.py b/pyanaconda/modules/payloads/payload/dnf/dnf_manager.py
index 823569185c6..f54039183be 100644
--- a/pyanaconda/modules/payloads/payload/dnf/dnf_manager.py
+++ b/pyanaconda/modules/payloads/payload/dnf/dnf_manager.py
@@ -18,6 +18,7 @@
 # Red Hat, Inc.
 #
 import multiprocessing
+import re
 import shutil
 import threading
 import traceback
@@ -582,6 +583,20 @@ def resolve_selection(self):
         log.info("The software selection has been resolved (%d packages selected).",
                  len(self._base.transaction))
 
+    def get_flatpak_refs(self):
+        """Determine what Flatpaks need to be preinstalled based on resolved transaction"""
+        if self._base.transaction is None:
+            return []
+
+        refs = []
+        for tsi in self._base.transaction:
+            for provide in tsi.pkg.provides:
+                m = re.match(r"^flatpak-preinstall\((.*)\)$", str(provide))
+                if m:
+                    refs.append(m.group(1))
+
+        return refs
+
     def clear_selection(self):
         """Clear the software selection."""
         self._base.reset(goal=True)
diff --git a/pyanaconda/modules/payloads/payload/factory.py b/pyanaconda/modules/payloads/payload/factory.py
index 00c69e0e2b5..34278d8476c 100644
--- a/pyanaconda/modules/payloads/payload/factory.py
+++ b/pyanaconda/modules/payloads/payload/factory.py
@@ -49,6 +49,10 @@ def create_payload(payload_type: PayloadType):
             from pyanaconda.modules.payloads.payload.rpm_ostree.rpm_ostree import RPMOSTreeModule
             return RPMOSTreeModule()
 
+        if payload_type == PayloadType.FLATPAK:
+            from pyanaconda.modules.payloads.payload.flatpak.flatpak import FlatpakModule
+            return FlatpakModule()
+
         raise ValueError("Unknown payload type: {}".format(payload_type))
 
     @classmethod
diff --git a/pyanaconda/modules/payloads/payload/flatpak/Makefile.am b/pyanaconda/modules/payloads/payload/flatpak/Makefile.am
new file mode 100644
index 00000000000..06a09ec2715
--- /dev/null
+++ b/pyanaconda/modules/payloads/payload/flatpak/Makefile.am
@@ -0,0 +1,21 @@
+#
+# Copyright (C) 2019  Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published
+# by the Free Software Foundation; either version 2.1 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+pkgpyexecdir = $(pyexecdir)/py$(PACKAGE_NAME)
+flatpak_moduledir = $(pkgpyexecdir)/modules/payloads/payload/flatpak
+dist_flatpak_module_DATA = $(wildcard $(srcdir)/*.py)
+
+MAINTAINERCLEANFILES = Makefile.in
diff --git a/pyanaconda/modules/payloads/payload/flatpak/__init__.py b/pyanaconda/modules/payloads/payload/flatpak/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/pyanaconda/modules/payloads/payload/flatpak/flatpak.py b/pyanaconda/modules/payloads/payload/flatpak/flatpak.py
new file mode 100644
index 00000000000..c644af38a4a
--- /dev/null
+++ b/pyanaconda/modules/payloads/payload/flatpak/flatpak.py
@@ -0,0 +1,112 @@
+#
+# Kickstart module for Live OS payload.
+#
+# Copyright (C) 2019 Red Hat, Inc.
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# the GNU General Public License v.2, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY expressed or implied, including the implied warranties of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
+# Public License for more details.  You should have received a copy of the
+# GNU General Public License along with this program; if not, write to the
+# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
+# source code or documentation are not subject to the GNU General Public
+# License and may only be used or replicated with the express permission of
+# Red Hat, Inc.
+#
+from pyanaconda.core.configuration.anaconda import conf
+from pyanaconda.modules.common.errors.payload import IncompatibleSourceError
+from pyanaconda.modules.payloads.base.utils import calculate_required_space
+from pyanaconda.modules.payloads.constants import SourceType, PayloadType
+from pyanaconda.modules.payloads.payload.flatpak.flatpak_manager import FlatpakManager
+from pyanaconda.modules.payloads.payload.flatpak.installation import CalculateFlatpaksSizeTask, \
+    CleanUpDownloadLocationTask, DownloadFlatpaksTask, InstallFlatpaksTask, \
+    PrepareDownloadLocationTask
+from pyanaconda.modules.payloads.payload.payload_base import PayloadBase
+from pyanaconda.modules.payloads.payload.flatpak.flatpak_interface import FlatpakInterface
+
+from pyanaconda.anaconda_loggers import get_module_logger
+log = get_module_logger(__name__)
+
+
+class FlatpakModule(PayloadBase):
+    """The Flatpak payload module."""
+
+    def __init__(self):
+        super().__init__()
+        self._flatpak_manager = FlatpakManager()
+
+    def for_publication(self):
+        """Get the interface used to publish this source."""
+        return FlatpakInterface(self)
+
+    @property
+    def type(self):
+        """Type of this payload."""
+        return PayloadType.FLATPAK
+
+    @property
+    def default_source_type(self):
+        """Type of the default source."""
+        return None
+
+    @property
+    def supported_source_types(self):
+        """List of supported source types."""
+        # Include all the types of SourceType.
+        return list(SourceType)
+
+    def set_sources(self, sources):
+        """Set a new list of sources to this payload.
+
+        This overrides the base implementation since the sources we set here
+        are the sources from the main payload, and can already be initialized.
+
+        :param sources: set a new sources
+        :type sources: instance of pyanaconda.modules.payloads.source.source_base.PayloadSourceBase
+        """
+        self._sources = sources
+        self._flatpak_manager.set_sources(sources)
+        self.sources_changed.emit()
+
+    def set_flatpak_refs(self, refs):
+        """Set the flatpak refs.
+
+        :param refs: a list of flatpak refs
+        """
+        self._flatpak_manager.set_flatpak_refs(refs)
+
+    def calculate_required_space(self):
+        """Calculate space required for the installation.
+
+        :return: required size in bytes
+        :rtype: int
+        """
+        return calculate_required_space(self._flatpak_manager.download_size,
+                                        self._flatpak_manager.install_size)
+
+    def install_with_tasks(self):
+        """Install the payload with tasks."""
+
+        tasks = [
+            CalculateFlatpaksSizeTask(
+                flatpak_manager=self._flatpak_manager,
+            ),
+            PrepareDownloadLocationTask(
+                flatpak_manager=self._flatpak_manager,
+            ),
+            DownloadFlatpaksTask(
+                flatpak_manager=self._flatpak_manager,
+            ),
+            InstallFlatpaksTask(
+                flatpak_manager=self._flatpak_manager,
+            ),
+            CleanUpDownloadLocationTask(
+                flatpak_manager=self._flatpak_manager,
+            ),
+        ]
+
+        return tasks
diff --git a/pyanaconda/modules/payloads/payload/flatpak/flatpak_interface.py b/pyanaconda/modules/payloads/payload/flatpak/flatpak_interface.py
new file mode 100644
index 00000000000..2904b829a1e
--- /dev/null
+++ b/pyanaconda/modules/payloads/payload/flatpak/flatpak_interface.py
@@ -0,0 +1,28 @@
+#
+# DBus interface for Flatpak payload.
+#
+# Copyright (C) 2024 Red Hat, Inc.
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# the GNU General Public License v.2, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY expressed or implied, including the implied warranties of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
+# Public License for more details.  You should have received a copy of the
+# GNU General Public License along with this program; if not, write to the
+# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
+# source code or documentation are not subject to the GNU General Public
+# License and may only be used or replicated with the express permission of
+# Red Hat, Inc.
+#
+from dasbus.server.interface import dbus_interface
+
+from pyanaconda.modules.common.constants.interfaces import PAYLOAD_FLATPAK
+from pyanaconda.modules.payloads.payload.payload_base_interface import PayloadBaseInterface
+
+
+@dbus_interface(PAYLOAD_FLATPAK.interface_name)
+class FlatpakInterface(PayloadBaseInterface):
+    """DBus interface for Flatpak payload module."""
diff --git a/pyanaconda/modules/payloads/payload/flatpak/flatpak_manager.py b/pyanaconda/modules/payloads/payload/flatpak/flatpak_manager.py
new file mode 100644
index 00000000000..545635905eb
--- /dev/null
+++ b/pyanaconda/modules/payloads/payload/flatpak/flatpak_manager.py
@@ -0,0 +1,279 @@
+#
+# Root object for handling Flatpak preinstallation
+#
+# Copyright (C) 2024 Red Hat, Inc.
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# the GNU General Public License v.2, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY expressed or implied, including the implied warranties of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
+# Public License for more details.  You should have received a copy of the
+# GNU General Public License along with this program; if not, write to the
+# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
+# source code or documentation are not subject to the GNU General Public
+# License and may only be used or replicated with the express permission of
+# Red Hat, Inc.
+#
+
+
+import os
+from typing import List, Optional
+
+from pyanaconda.anaconda_loggers import get_module_logger
+from pyanaconda.core.configuration.anaconda import conf
+from pyanaconda.core.glib import GError
+from pyanaconda.core.i18n import _
+from pyanaconda.modules.common.errors.installation import PayloadInstallationError
+from pyanaconda.modules.common.structures.payload import RepoConfigurationData
+from pyanaconda.modules.common.task.progress import ProgressReporter
+from pyanaconda.modules.payloads.constants import SourceType
+from pyanaconda.modules.payloads.payload.flatpak.source import FlatpakRegistrySource, \
+    FlatpakStaticSource, NoSourceError
+from pyanaconda.modules.payloads.source.source_base import PayloadSourceBase, RepositorySourceMixin
+
+import gi
+gi.require_version("Flatpak", "1.0")
+gi.require_version("Gio", "2.0")
+
+from gi.repository.Flatpak import Transaction, Installation, TransactionOperationType
+from gi.repository.Gio import File
+
+
+# We need Flatpak to read configuration files from the target and write
+# to the target system installation. Since we use the Flatpak API
+# in process, we need to do this by modifying the environment before
+# we start any threads. Setting these variables will be harmless if
+# we aren't actually using Flatpak.
+
+# pylint: disable=environment-modify
+os.environ["FLATPAK_DOWNLOAD_TMPDIR"] = os.path.join(conf.target.system_root, "var/tmp")
+# pylint: disable=environment-modify
+os.environ["FLATPAK_CONFIG_DIR"] = os.path.join(conf.target.system_root, "etc/flatpak")
+# pylint: disable=environment-modify
+os.environ["FLATPAK_OS_CONFIG_DIR"] = os.path.join(conf.target.system_root, "usr/share/flatpak")
+# pylint: disable=environment-modify
+os.environ["FLATPAK_SYSTEM_DIR"] = os.path.join(conf.target.system_root, "var/lib/flatpak")
+
+
+log = get_module_logger(__name__)
+
+__all__ = ["FlatpakManager"]
+
+
+class FlatpakManager:
+    """Root object for handling Flatpak preinstallation"""
+
+    def __init__(self):
+        """Create and initialize this class.
+
+        :param function callback: a progress reporting callback
+        """
+        self._flatpak_refs = []
+        self._source_repository = None
+        self._source = None
+        self._skip_installation = False
+        self._collection_location = None
+        self._progress: Optional[ProgressReporter] = None
+        self._transaction = None
+
+    def set_sources(self, sources: List[PayloadSourceBase]):
+        """Set the source object we use to download Flatpak content.
+
+        If unset, preinstallation will install directly from the configured
+        Flatpak remote (see flatpak_remote in the anaconda configuration).
+
+        :param str url: URL pointing to the Flatpak content
+        """
+
+        source = sources[0]
+
+        if isinstance(source, RepositorySourceMixin):
+            if self._source and isinstance(self._source, FlatpakStaticSource) \
+                    and self._source.repository_config == source.repository:
+                return
+            self._source = FlatpakStaticSource(source.repository, relative_path="Flatpaks")
+        elif source.type in (SourceType.CDN, SourceType.CLOSEST_MIRROR):
+            if self._source and isinstance(self._source, FlatpakRegistrySource):
+                return
+            _, remote_url = conf.payload.flatpak_remote
+            log.debug("Using Flatpak registry source: %s", remote_url)
+            self._source = FlatpakRegistrySource(remote_url)
+        else:
+            self._source = None
+
+    def set_flatpak_refs(self, refs: Optional[List[str]]):
+        """Set the Flatpak refs to be installed.
+
+        :param refs: List of Flatpak refs to be installed, None to use
+           all Flatpak refs from the source. Each ref should be in the form
+           [<collection_id>:](app|runtime)/<id>/[<arch>]/<branch>
+        """
+        self._skip_installation = False
+        self._flatpak_refs = refs if refs is not None else []
+
+    def set_download_location(self, path: str):
+        """Sets a location that can be used for temporary download of Flatpak content.
+
+        :param path: parent directory to store downloaded Flatpak content
+           (the download should be to a subdirectory of this path)
+        """
+        self._download_location = path
+
+    @property
+    def download_location(self) -> str:
+        """Get the download location."""
+        return self._download_location
+
+    def _get_source(self):
+        if self._source is None:
+            if self._source_repository:
+                log.debug("Using Flatpak source repository at: %s/Flatpaks",
+                          self._source_repository.url)
+                self._source = FlatpakStaticSource(self._source_repository,
+                                                   relative_path="Flatpaks")
+            else:
+                _, remote_url = conf.payload.flatpak_remote
+                log.debug("Using Flatpak registry source: %s", remote_url)
+                self._source = FlatpakRegistrySource(remote_url)
+
+        return self._source
+
+    def calculate_size(self, progress: Optional[ProgressReporter]):
+        """Calculate the download and install size of the Flatpak content.
+
+        :param progress: used to report progress of the operation
+
+        The result is available from the download_size and install_size properties.
+        """
+        if self._skip_installation or len(self._flatpak_refs) == 0:
+            return
+
+        try:
+            self._download_size, self._install_size = \
+                self._get_source().calculate_size(self._flatpak_refs)
+        except NoSourceError as e:
+            log.info("Flatpak source not available, skipping: %s", e)
+            self._skip_installation = True
+
+    @property
+    def download_size(self):
+        """Space needed to to temporarily download Flatpak content before installation"""
+        return self._download_size
+
+    @property
+    def install_size(self):
+        """Space used after installation in the target system"""
+        return self._install_size
+
+    def download(self, progress: ProgressReporter):
+        """Download Flatpak content to a temporary location.
+
+        :param progress: used to report progress of the operation
+
+        This is only needed if Flatpak can't install the content directly.
+        """
+        if self._skip_installation or len(self._flatpak_refs) == 0:
+            return
+
+        try:
+            self._collection_location = self._get_source().download(self._flatpak_refs,
+                                                                    self._download_location,
+                                                                    progress)
+        except NoSourceError as e:
+            log.info("Flatpak source not available, skipping: %s", e)
+            self._skip_installation = True
+
+    def install(self, progress: ProgressReporter):
+        """Install the Flatpak content to the target system.
+
+        :param progress: used to report progress of the operation
+        """
+        if self._skip_installation or len(self._flatpak_refs) == 0:
+            return
+
+        installation = self._create_flatpak_installation()
+        self._transaction = self._create_flatpak_transaction(installation)
+
+        if self._collection_location:
+            self._transaction.add_sideload_image_collection(self._collection_location, None)
+
+        self._transaction.add_sync_preinstalled()
+
+        try:
+            self._progress = progress
+            self._transaction.run()
+        except GError as e:
+            raise PayloadInstallationError("Failed to install flatpaks: {}".format(e)) from e
+        finally:
+            self._transaction.run_dispose()
+            self._transaction = None
+            self._progress = None
+
+    def _create_flatpak_installation(self):
+        return Installation.new_system(None)
+
+    def _create_flatpak_transaction(self, installation):
+        transaction = Transaction.new_for_installation(installation)
+        transaction.connect("new_operation", self._operation_started_callback)
+        transaction.connect("operation_done", self._operation_stopped_callback)
+        transaction.connect("operation_error", self._operation_error_callback)
+
+        return transaction
+
+    def _operation_started_callback(self, transaction, operation, progress):
+        """Start of the new operation.
+
+        :param transaction: the main transaction object
+        :type transaction: Flatpak.Transaction instance
+        :param operation: object describing the operation
+        :type operation: Flatpak.TransactionOperation instance
+        :param progress: object providing progess of the operation
+        :type progress: Flatpak.TransactionProgress instance
+        """
+        self._log_operation(operation, "started")
+        self._report_progress(_("Installing {}").format(operation.get_ref()))
+
+    def _operation_stopped_callback(self, transaction, operation, _commit, result):
+        """Existing operation ended.
+
+        :param transaction: the main transaction object
+        :type transaction: Flatpak.Transaction instance
+        :param operation: object describing the operation
+        :type operation: Flatpak.TransactionOperation instance
+        :param str _commit: operation was committed this is a commit id
+        :param result: object containing details about the result of the operation
+        :type result: Flatpak.TransactionResult instance
+        """
+        self._log_operation(operation, "stopped")
+
+    def _operation_error_callback(self, transaction, operation, error, details):
+        """Process error raised by the flatpak operation.
+
+        :param transaction: the main transaction object
+        :type transaction: Flatpak.Transaction instance
+        :param operation: object describing the operation
+        :type operation: Flatpak.TransactionOperation instance
+        :param error: object containing error description
+        :type error: GLib.Error instance
+        :param details: information if the error was fatal
+        :type details: int value of Flatpak.TransactionErrorDetails
+        """
+        self._log_operation(operation, "failed")
+        log.error("Flatpak operation has failed with a message: '%s'", error.message)
+
+    def _report_progress(self, message):
+        """Report a progress message."""
+        if not self._progress:
+            return
+
+        self._progress.report_progress(message)
+
+    @staticmethod
+    def _log_operation(operation, state):
+        """Log a Flatpak operation."""
+        operation_type_str = TransactionOperationType.to_string(operation.get_operation_type())
+        log.debug("Flatpak operation: %s of ref %s state %s",
+                  operation_type_str, operation.get_ref(), state)
diff --git a/pyanaconda/modules/payloads/payload/flatpak/installation.py b/pyanaconda/modules/payloads/payload/flatpak/installation.py
new file mode 100644
index 00000000000..350de705512
--- /dev/null
+++ b/pyanaconda/modules/payloads/payload/flatpak/installation.py
@@ -0,0 +1,144 @@
+#
+# Copyright (C) 2024 Red Hat, Inc.
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# the GNU General Public License v.2, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY expressed or implied, including the implied warranties of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
+# Public License for more details.  You should have received a copy of the
+# GNU General Public License along with this program; if not, write to the
+# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
+# source code or documentation are not subject to the GNU General Public
+# License and may only be used or replicated with the express permission of
+# Red Hat, Inc.
+#
+
+import os
+import shutil
+from pyanaconda.anaconda_loggers import get_module_logger
+from pyanaconda.modules.common.task import Task
+from pyanaconda.modules.payloads.base.utils import pick_download_location
+from pyanaconda.modules.payloads.payload.flatpak.flatpak_manager import FlatpakManager
+
+log = get_module_logger(__name__)
+
+FLATPAK_MIRROR_DIR_SUFFIX = 'flatpak.mirror'
+
+
+class CalculateFlatpaksSizeTask(Task):
+    """Task to determine space needed for Flatpaks"""
+
+    def __init__(self, flatpak_manager: FlatpakManager):
+        """Create a new task."""
+        super().__init__()
+        self._flatpak_manager = flatpak_manager
+
+    @property
+    def name(self):
+        """Name of the task."""
+        return "Calculate needed space for Flatpaks"
+
+    def run(self):
+        self._flatpak_manager.calculate_size(self)
+
+
+class PrepareDownloadLocationTask(Task):
+    """The installation task for setting up the download location."""
+
+    def __init__(self, flatpak_manager: FlatpakManager):
+        """Create a new task.
+
+        :param dnf_manager: a DNF manager
+        """
+        super().__init__()
+        self._flatpak_manager = flatpak_manager
+
+    @property
+    def name(self):
+        return "Prepare the package download"
+
+    def run(self):
+        """Run the task.
+
+        :return: a path of the download location
+        """
+
+        self._flatpak_manager.calculate_size(self)
+
+        path = pick_download_location(self._flatpak_manager.download_size,
+                                      self._flatpak_manager.install_size,
+                                      FLATPAK_MIRROR_DIR_SUFFIX)
+
+        if os.path.exists(path):
+            log.info("Removing existing package download location: %s", path)
+            shutil.rmtree(path)
+
+        self._flatpak_manager.set_download_location(path)
+        return path
+
+
+class CleanUpDownloadLocationTask(Task):
+    """The installation task for cleaning up the download location."""
+
+    def __init__(self, flatpak_manager):
+        """Create a new task.
+
+        :param flatpak_manager: a Flatpak manager
+        """
+        super().__init__()
+        self._flatpak_manager = flatpak_manager
+
+    @property
+    def name(self):
+        return "Remove downloaded Flatpaks"
+
+    def run(self):
+        """Run the task.
+        """
+        path = self._flatpak_manager.download_location
+
+        if not os.path.exists(path):
+            # If nothing was downloaded, there is nothing to clean up.
+            return
+
+        log.info("Removing downloaded packages from %s.", path)
+        shutil.rmtree(path)
+
+
+class DownloadFlatpaksTask(Task):
+    """Task to download remote Flatpaks"""
+
+    def __init__(self, flatpak_manager):
+        """Create a new task."""
+        super().__init__()
+        self._flatpak_manager = flatpak_manager
+
+    @property
+    def name(self):
+        """Name of the task."""
+        return "Download remote Flatpaks"
+
+    def run(self):
+        """Run the task."""
+        self._flatpak_manager.download(self)
+
+
+class InstallFlatpaksTask(Task):
+    """Task to install flatpaks"""
+
+    def __init__(self, flatpak_manager):
+        """Create a new task."""
+        super().__init__()
+        self._flatpak_manager = flatpak_manager
+
+    @property
+    def name(self):
+        """Name of the task."""
+        return "Install flatpaks"
+
+    def run(self):
+        """Run the task."""
+        self._flatpak_manager.install(self)
diff --git a/pyanaconda/modules/payloads/payload/flatpak/source.py b/pyanaconda/modules/payloads/payload/flatpak/source.py
new file mode 100644
index 00000000000..3e2712974f5
--- /dev/null
+++ b/pyanaconda/modules/payloads/payload/flatpak/source.py
@@ -0,0 +1,434 @@
+#
+# Query and download sources of Flatpak content
+#
+# Copyright (C) 2024 Red Hat, Inc.
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# the GNU General Public License v.2, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY expressed or implied, including the implied warranties of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
+# Public License for more details.  You should have received a copy of the
+# GNU General Public License along with this program; if not, write to the
+# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
+# source code or documentation are not subject to the GNU General Public
+# License and may only be used or replicated with the express permission of
+# Red Hat, Inc.
+#
+
+from abc import ABC, abstractmethod
+from configparser import ConfigParser, NoSectionError
+from contextlib import contextmanager
+from functools import cached_property
+import json
+import os
+from typing import Dict, Generator, List, Optional, Tuple
+from urllib.parse import urljoin, urlparse
+
+from blivet.arch import get_arch
+import requests
+
+from pyanaconda.anaconda_loggers import get_module_logger
+from pyanaconda.core.i18n import _
+from pyanaconda.core.util import requests_session
+from pyanaconda.modules.common.structures.payload import RepoConfigurationData
+from pyanaconda.modules.common.task.progress import ProgressReporter
+from pyanaconda.modules.payloads.base.utils import get_downloader_for_repo_configuration
+
+log = get_module_logger(__name__)
+
+__all__ = ["FlatpakSource", "FlatpakStaticSource", "FlatpakRegistrySource", "NoSourceError"]
+
+
+_CONTAINER_ARCH_MAP = {
+    "x86_64": "amd64",
+    "aarch64": "arm64"
+}
+
+
+def _get_container_arch():
+    """Architecture name as used by docker/podman"""
+    arch = get_arch()
+    return _CONTAINER_ARCH_MAP.get(arch, arch)
+
+
+def _canonicalize_flatpak_ref(ref) -> Tuple[Optional[str], str]:
+    """Split off a collection ID, and add architecture if unspecified
+
+    Turn "org.fedoraproject.Stable:app/org.example.Foo//stable" into
+    ("org.fedoraproject.Stable", "app/org.example.Foo/amd64/stable")
+    """
+
+    collection_parts = ref.split(":", 1)
+    if len(collection_parts) == 2:
+        collection = collection_parts[0]
+        ref = collection_parts[1]
+    else:
+        collection = None
+
+    parts = ref.split("/")
+    if len(parts) != 4:
+        raise RuntimeError("Can't parse reference")
+    if parts[2] == "":
+        parts[2] = get_arch()
+
+    return collection, "/".join(parts)
+
+
+class NoSourceError(Exception):
+    """Source not found."""
+
+
+class SourceImage(ABC):
+    """Representation of a single image of a FlatpakSource."""
+
+    @property
+    @abstractmethod
+    def labels(self) -> Dict[str, str]:
+        """The labels of the image."""
+
+    @property
+    def ref(self) -> Optional[str]:
+        """Flatpak reference for the image, or None if not a Flatpak"""
+        return self.labels.get("org.flatpak.ref")
+
+    @property
+    def download_size(self) -> int:
+        """Download size, in bytes"""
+        return int(self.labels["org.flatpak.download-size"])
+
+    @property
+    def installed_size(self) -> int:
+        """Installed size, in bytes"""
+        return int(self.labels["org.flatpak.installed-size"])
+
+
+class FlatpakSource(ABC):
+    """Base class for places where Flatpak images can be downloaded from."""
+
+    @abstractmethod
+    def calculate_size(self, refs: List[str]) -> Tuple[int, int]:
+        """Calculate the total download and installed size of the images in refs and
+        their dependencies.
+
+        :param refs: list of Flatpak references
+        :returns: download size, installed size
+        """
+
+    @abstractmethod
+    def download(self, refs: List[str], download_location: str,
+                 progress: Optional[ProgressReporter] = None) -> Optional[str]:
+        """Downloads the images referenced by refs and any dependencies.
+
+        If they are already local, or they can be installed
+        directly from the remote location, nothing will be downloaded.
+
+        Whether or not anything as been downloaded, returns
+        the specification of a sideload repository that can be used to install from
+        this source, or None if none is needed.
+
+        :param refs: list of Flatpak references
+        :param download_location: path to location for temporary downloads
+        :param progress: used to report progress of the download
+        :returns sideload location, including the transport (e.g. oci:<path>), or None
+        """
+
+    @property
+    @abstractmethod
+    def _images(self) -> List[SourceImage]:
+        """All images in the source, filtered for the current architecture."""
+        ...
+
+    def _expand_refs(self, refs: List[str]) -> List[str]:
+        """Expand the list of refs to be in full form and include any dependencies."""
+        result = []
+        for ref in refs:
+            # We don't do anything with the collection ID for now
+            _, ref = _canonicalize_flatpak_ref(ref)
+            result.append(ref)
+
+        for image in self._images:
+            if image.ref not in result:
+                continue
+
+            metadata = image.labels.get("org.flatpak.metadata")
+            if metadata is None:
+                continue
+
+            cp = ConfigParser(interpolation=None)
+            cp.read_string(metadata)
+            try:
+                runtime = cp.get('Application', 'Runtime')
+                if runtime:
+                    runtime_ref = f"runtime/{runtime}"
+                    if runtime_ref not in result:
+                        result.append(runtime_ref)
+            except (NoSectionError, KeyError):
+                pass
+
+        return result
+
+
+class StaticSourceImage(SourceImage):
+    """One image of a FlatpakStaticSource."""
+
+    def __init__(self, digest, manifest_json, config_json):
+        self.digest = digest
+        self.manifest_json = manifest_json
+        self.config_json = config_json
+
+    @property
+    def labels(self):
+        return self.config_json["config"]["Labels"]
+
+    @property
+    def download_size(self):
+        # This is more accurate than using the org.flatpak.download-size label,
+        # because further processing of the image might have recompressed
+        # the layer using different settings.
+        return sum(int(layer["size"]) for layer in self.manifest_json["layers"])
+
+
+class FlatpakStaticSource(FlatpakSource):
+    """Flatpak images stored in a OCI image layout, either locally or remotely
+
+    https://github.com/opencontainers/image-spec/blob/main/image-layout.md
+    """
+
+    def __init__(self, repository_config: RepoConfigurationData, relative_path: str = "Flatpaks"):
+        """Create a new source.
+
+        :param repository_config: URL of the repository, or a local path
+        :param relative_path: path of an OCI layout, relative to the repository root
+        """
+        self.repository_config = repository_config
+        self._url = urljoin(repository_config.url + "/", relative_path)
+        self._is_local = self._url.startswith("file://")
+        self._cached_blobs = {}
+
+    @contextmanager
+    def _downloader(self):
+        """Prepare a requests.Session.get method appropriately for the repository.
+
+        :returns: a function that acts like requests.Session.get()
+        """
+        with requests_session() as session:
+            downloader = get_downloader_for_repo_configuration(session, self.repository_config)
+            yield downloader
+
+    def calculate_size(self, refs):
+        """Calculate the total download and installed size of the images in refs and
+        their dependencies.
+
+        :param refs: list of Flatpak references
+        :returns: download size, installed size
+        """
+        log.debug("Calculating size of: %s", refs)
+
+        download_size = 0
+        installed_size = 0
+        expanded = self._expand_refs(refs)
+
+        for image in self._images:
+            if image.ref not in expanded:
+                continue
+
+            log.debug("%s: download %d%s, installed %d",
+                      image.ref,
+                      " (skipped)" if self._is_local else "",
+                      image.download_size, image.installed_size)
+            download_size += 0 if self._is_local else image.download_size
+            installed_size += image.installed_size
+
+        log.debug("Total: download %d, installed %d", download_size, installed_size)
+        return download_size, installed_size
+
+    def download(self, refs, download_location, progress=None):
+        if self._is_local:
+            return "oci:" + self._url.removeprefix("file://")
+
+        collection_location = os.path.join(download_location, "Flatpaks")
+        expanded_refs = self._expand_refs(refs)
+
+        index_json = {
+            "schemaVersion": 2,
+            "manifests": []
+        }
+
+        with self._downloader() as downloader:
+            for image in self._images:
+                if image.ref in expanded_refs:
+                    log.debug("Downloading %s, %s bytes", image.ref, image.download_size)
+                    if progress:
+                        progress.report_progress(_("Downloading {}").format(image.ref))
+
+                    manifest_len = self._download_blob(downloader, download_location, image.digest)
+                    self._download_blob(downloader,
+                                        download_location, image.manifest_json["config"]["digest"])
+                    index_json["manifests"].append({
+                        "mediaType": "application/vnd.oci.image.manifest.v1+json",
+                        "digest": image.digest,
+                        "size": manifest_len
+                    })
+
+                    for layer in image.manifest_json["layers"]:
+                        self._download_blob(downloader,
+                                            download_location, layer["digest"],
+                                            stream=True)
+
+        os.makedirs(collection_location, exist_ok=True)
+        with open(os.path.join(collection_location, "index.json"), "w") as f:
+            json.dump(index_json, f)
+
+        with open(os.path.join(collection_location, "oci-layout"), "w") as f:
+            json.dump({
+                "imageLayoutVersion": "1.0.0"
+            }, f)
+
+        return "oci:" + collection_location
+
+    @cached_property
+    def _images(self) -> List[StaticSourceImage]:
+        result = []
+
+        with self._downloader() as downloader:
+            url = self._url + "/index.json"
+            response = downloader(url)
+            if response.status_code == 404:
+                raise NoSourceError("No source found at {}".format(url))
+            response.raise_for_status()
+            index_json = response.json()
+
+            for manifest in index_json.get("manifests", ()):
+                if manifest.get("mediaType") == "application/vnd.oci.image.manifest.v1+json":
+                    digest = manifest["digest"]
+                    manifest_json = self._get_json(downloader, manifest["digest"])
+                    config_json = self._get_json(downloader, manifest_json["config"]["digest"])
+                    result.append(StaticSourceImage(digest, manifest_json, config_json))
+
+        return result
+
+    def _blob_url(self, digest):
+        assert digest.startswith("sha256:")
+        return self._url + "/blobs/sha256/" + digest[7:]
+
+    def _get_blob(self, downloader, digest) -> bytes:
+        result = self._cached_blobs.get(digest)
+        if result:
+            return result
+
+        response = downloader(self._blob_url(digest))
+        response.raise_for_status()
+
+        self._cached_blobs[digest] = result = response.content
+        return result
+
+    def _download_blob(self, downloader, download_location, digest, stream=False):
+        assert digest.startswith("sha256:")
+
+        blobs_dir = os.path.join(download_location, "blobs/sha256/")
+        os.makedirs(blobs_dir, exist_ok=True)
+
+        path = os.path.join(blobs_dir, digest[7:])
+        with open(path, "wb") as f:
+            if stream:
+                response = downloader(self._blob_url(digest), stream=True)
+                response.raise_for_status()
+                size = 0
+                while True:
+                    chunk = response.raw.read(64*1024)
+                    if not chunk:
+                        break
+                    size += len(chunk)
+                    f.write(chunk)
+                return size
+            else:
+                blob = self._get_blob(downloader, digest)
+                f.write(blob)
+                return len(blob)
+
+    def _get_json(self, session, digest):
+        return json.loads(self._get_blob(session, digest))
+
+
+class RegistrySourceImage(SourceImage):
+    def __init__(self, labels):
+        self._labels = labels
+
+    @property
+    def labels(self):
+        return self._labels
+
+
+class FlatpakRegistrySource(FlatpakSource):
+    """Flatpak images indexed by a remote JSON file, and stored in a registry.
+
+    https://github.com/flatpak/flatpak-oci-specs/blob/main/registry-index.md
+    """
+
+    def __init__(self, url):
+        self._index = None
+        self._url = url
+
+    def calculate_size(self, refs):
+        # For registry sources, we don't download the images in advance;
+        # instead they are downloaded into the /var/tmp of the target
+        # system and installed one-by-one. So the downloads don't count
+        # towards the space in the temporary download location, but we
+        # need space for the largest download in the target system.
+        # (That space will also be needed for upgrades after installation.)
+
+        log.debug("Calculating size of: %s", refs)
+
+        max_download_size = 0
+        installed_size = 0
+        expanded = self._expand_refs(refs)
+
+        for image in self._images:
+            if image.ref not in expanded:
+                continue
+
+            log.debug("%s: download %d, installed %d",
+                      image.ref, image.download_size, image.installed_size)
+
+            max_download_size = max(max_download_size, image.download_size)
+            installed_size += image.installed_size
+
+        log.debug("Total: max download %d, installed %d", max_download_size, installed_size)
+        return 0, installed_size + max_download_size
+
+    @cached_property
+    def _images(self):
+        arch = _get_container_arch()
+
+        base_url = self._url.removeprefix("oci+")
+        parsed = urlparse(base_url)
+        if parsed.fragment:
+            tag = parsed.fragment
+            base_url = parsed._replace(fragment=None, query=None).geturl()
+        else:
+            tag = "latest"
+
+        url_pattern = "{}/index/static?label:org.flatpak.ref:exists=1&architecture={}&tag={}"
+        full_url = url_pattern.format(base_url, arch, tag)
+        with requests_session() as session:
+            response = session.get(full_url)
+            response.raise_for_status()
+            index = response.json()
+
+        result = []
+
+        arch = _get_container_arch()
+        for repository in index["Results"]:
+            for image in repository["Images"]:
+                if image['Architecture'] != arch:
+                    continue
+
+                result.append(RegistrySourceImage(image["Labels"]))
+
+        return result
+
+    def download(self, refs, download_location, progress=None):
+        return None
diff --git a/pyanaconda/modules/payloads/payload/payload_base.py b/pyanaconda/modules/payloads/payload/payload_base.py
index 6a12c1cb680..22ebc641205 100644
--- a/pyanaconda/modules/payloads/payload/payload_base.py
+++ b/pyanaconda/modules/payloads/payload/payload_base.py
@@ -179,6 +179,20 @@ def set_kernel_version_list(self, kernels):
         self._kernel_version_list = kernels
         log.debug("The kernel version list is set to: %s", kernels)
 
+    def needs_flatpak_side_payload(self):
+        """Does this payload need an extra payload for Flatpak installation
+
+        :return: True or False
+        """
+        return False
+
+    def get_flatpak_refs(self):
+        """Get the list of Flatpak refs to install.
+
+        :return: list of Flatpak refs
+        """
+        return []
+
     @abstractmethod
     def install_with_tasks(self):
         """Install the payload.
diff --git a/pyanaconda/modules/payloads/payloads.py b/pyanaconda/modules/payloads/payloads.py
index 8060fdee161..aacac9330f1 100644
--- a/pyanaconda/modules/payloads/payloads.py
+++ b/pyanaconda/modules/payloads/payloads.py
@@ -24,10 +24,12 @@
 from pyanaconda.modules.common.base import KickstartService
 from pyanaconda.modules.common.constants.services import PAYLOADS
 from pyanaconda.modules.common.containers import TaskContainer
+from pyanaconda.modules.payloads.constants import PayloadType
 from pyanaconda.modules.payloads.installation import PrepareSystemForInstallationTask, \
     CopyDriverDisksFilesTask
 from pyanaconda.modules.payloads.kickstart import PayloadKickstartSpecification
 from pyanaconda.modules.payloads.payload.factory import PayloadFactory
+from pyanaconda.modules.payloads.payload.flatpak.flatpak import FlatpakModule
 from pyanaconda.modules.payloads.payloads_interface import PayloadsInterface
 from pyanaconda.modules.payloads.source.factory import SourceFactory
 
@@ -47,6 +49,8 @@ def __init__(self):
         self._active_payload = None
         self.active_payload_changed = Signal()
 
+        self._flatpak_side_payload = None
+
     def publish(self):
         """Publish the module."""
         TaskContainer.set_namespace(PAYLOADS.namespace)
@@ -87,6 +91,14 @@ def active_payload(self):
     def activate_payload(self, payload):
         """Activate the payload."""
         self._active_payload = payload
+
+        if self._active_payload.needs_flatpak_side_payload():
+            payload = self.create_payload(PayloadType.FLATPAK)
+            assert isinstance(payload, FlatpakModule)
+            self._flatpak_side_payload = payload
+        else:
+            self._flatpak_side_payload = None
+
         self.active_payload_changed.emit()
         log.debug("Activated the payload %s.", payload.type)
 
@@ -140,6 +152,10 @@ def calculate_required_space(self):
 
         if self.active_payload:
             total += self.active_payload.calculate_required_space()
+            if self._flatpak_side_payload:
+                self._flatpak_side_payload.set_sources(self.active_payload.get_sources())
+                self._flatpak_side_payload.set_flatpak_refs(self.active_payload.get_flatpak_refs())
+                total += self._flatpak_side_payload.calculate_required_space()
 
         return total
 
@@ -174,6 +190,12 @@ def install_with_tasks(self):
         ]
 
         tasks += self.active_payload.install_with_tasks()
+
+        if self._flatpak_side_payload:
+            self._flatpak_side_payload.set_sources(self.active_payload.sources)
+            self._flatpak_side_payload.set_flatpak_refs(self.active_payload.get_flatpak_refs())
+            tasks += self._flatpak_side_payload.install_with_tasks()
+
         return tasks
 
     def post_install_with_tasks(self):
@@ -191,6 +213,10 @@ def post_install_with_tasks(self):
         ]
 
         tasks += self.active_payload.post_install_with_tasks()
+
+        if self._flatpak_side_payload:
+            tasks += self._flatpak_side_payload.post_install_with_tasks()
+
         return tasks
 
     def teardown_with_tasks(self):
@@ -203,4 +229,7 @@ def teardown_with_tasks(self):
         if self.active_payload:
             tasks += self.active_payload.tear_down_with_tasks()
 
+        if self._flatpak_side_payload:
+            tasks += self._flatpak_side_payload.tear_down_with_tasks()
+
         return tasks