Feat(streaming): emulator streaming v2

backend/alembic/env.py

@@ -6,7 +6,13 @@
 
 from config.config_manager import ConfigManager
 from logger.logger import unify_logger
-from models.assets import Save, Screenshot, State  # noqa
+from models.assets import (  # noqa
+    MemoryCard,
+    MemoryCardVersion,
+    Save,
+    Screenshot,
+    State,
+)
 from models.base import BaseModel
 from models.collection import VirtualCollection
 from models.firmware import Firmware  # noqa

backend/alembic/versions/0108_memory_cards.py

@@ -0,0 +1,112 @@
+"""Add memory_cards and memory_card_versions tables
+
+Revision ID: 0108_memory_cards
+Revises: 0107_roms_dedup_cover_index
+Create Date: 2026-07-12 00:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0108_memory_cards"
+down_revision = "0107_roms_dedup_cover_index"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "memory_cards",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=False),
+        sa.Column("emulator", sa.String(length=50), nullable=False),
+        sa.Column("platform_id", sa.Integer(), nullable=True),
+        sa.Column("name", sa.String(length=255), nullable=False),
+        sa.Column("slot", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column(
+            "is_public",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.text("false"),
+        ),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["platform_id"], ["platforms.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    with op.batch_alter_table("memory_cards") as batch_op:
+        batch_op.create_index(
+            batch_op.f("ix_memory_cards_user_emulator"),
+            ["user_id", "emulator"],
+        )
+        batch_op.create_index(
+            batch_op.f("ix_memory_cards_public"),
+            ["is_public"],
+        )
+
+    op.create_table(
+        "memory_card_versions",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("memory_card_id", sa.Integer(), nullable=False),
+        sa.Column("file_name", sa.String(length=450), nullable=False),
+        sa.Column("file_name_no_tags", sa.String(length=450), nullable=False),
+        sa.Column("file_name_no_ext", sa.String(length=450), nullable=False),
+        sa.Column("file_extension", sa.String(length=100), nullable=False),
+        sa.Column("file_path", sa.String(length=1000), nullable=False),
+        sa.Column(
+            "file_size_bytes", sa.BigInteger(), nullable=False, server_default="0"
+        ),
+        sa.Column(
+            "missing_from_fs",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.text("false"),
+        ),
+        sa.Column("content_hash", sa.String(length=32), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["memory_card_id"], ["memory_cards.id"], ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    with op.batch_alter_table("memory_card_versions") as batch_op:
+        batch_op.create_index(
+            batch_op.f("ix_memory_card_versions_card"),
+            ["memory_card_id"],
+        )
+        batch_op.create_index(
+            batch_op.f("ix_memory_card_versions_card_hash"),
+            ["memory_card_id", "content_hash"],
+        )
+
+
+def downgrade() -> None:
+    # drop_table removes the tables' indexes and foreign keys; dropping the
+    # FK-backing indexes explicitly first is both redundant and rejected by
+    # MariaDB/MySQL.
+    op.drop_table("memory_card_versions")
+    op.drop_table("memory_cards")

backend/alembic/versions/0109_container_adoptions.py

@@ -0,0 +1,53 @@
+"""Add streaming_container_adoptions table
+
+Revision ID: 0109_container_adoptions
+Revises: 0108_memory_cards
+Create Date: 2026-07-21 00:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0109_container_adoptions"
+down_revision = "0108_memory_cards"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "streaming_container_adoptions",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("container_key", sa.String(length=512), nullable=False),
+        sa.Column("outcome", sa.String(length=16), nullable=False),
+        sa.Column("decided_by_user_id", sa.Integer(), nullable=True),
+        sa.Column(
+            "decided_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["decided_by_user_id"], ["users.id"], ondelete="SET NULL"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("container_key"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("streaming_container_adoptions")

backend/alembic/versions/0110_state_disc_file.py

@@ -0,0 +1,38 @@
+"""add disc_file_id to states
+
+Revision ID: 0110_state_disc_file
+Revises: 0109_container_adoptions
+Create Date: 2026-08-14 00:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0110_state_disc_file"
+down_revision = "0109_container_adoptions"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("states", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("disc_file_id", sa.Integer(), nullable=True))
+        # Postgres indexes no FK column on its own, and the SET NULL cascade
+        # scans this on every rom_files delete.
+        batch_op.create_index("ix_states_disc_file_id", ["disc_file_id"])
+        batch_op.create_foreign_key(
+            "fk_states_disc_file_id",
+            "rom_files",
+            ["disc_file_id"],
+            ["id"],
+            ondelete="SET NULL",
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("states", schema=None) as batch_op:
+        batch_op.drop_constraint("fk_states_disc_file_id", type_="foreignkey")
+        batch_op.drop_index("ix_states_disc_file_id")
+        batch_op.drop_column("disc_file_id")

backend/config/__init__.py

@@ -316,6 +316,12 @@ def _get_env(var: str, fallback: str | None = None) -> str | None:
 STREAMING_SAVE_TIMEOUT: Final[int] = safe_int(
     _get_env("STREAMING_SAVE_TIMEOUT"), 45
 )  # 45 seconds
+# How many save states to keep per ROM, emulator and user. Each capture is
+# kept as its own asset rather than overwriting a slot, so the oldest are
+# pruned once this many exist. 0 disables pruning.
+STREAMING_STATE_HISTORY_LIMIT: Final[int] = safe_int(
+    _get_env("STREAMING_STATE_HISTORY_LIMIT"), 50
+)
 
 # SENTRY
 SENTRY_DSN: Final[str | None] = _get_env("SENTRY_DSN")

backend/config/config_manager.py

@@ -162,11 +162,39 @@ class NetplayICEServer(TypedDict):
     credential: NotRequired[str]
 
 
+class StreamingPlatformOverride(TypedDict):
+    # Names the state and card namespace, so it has no container-level default.
+    emulator: str
+    # Anything set here wins over the same key on the container.
+    label: NotRequired[str]
+    memory_card_sync: NotRequired[bool]
+
+
 class StreamingContainer(TypedDict):
-    platform: str
+    # A container declares either one platform (the per-emulator mods) or a
+    # `platforms` map (one webstation serving many). Exactly one of the two.
+    platform: NotRequired[str]
+    # Platform slug to the emulator that serves it, or to a block of options
+    # overriding container keys for that platform, replacing platform +
+    # emulator on a container that hosts more than one.
+    platforms: NotRequired[dict[str, str | StreamingPlatformOverride]]
     host: str
-    broker_host: str
+    # Optional under `protocol: webstation`, which derives the broker host from
+    # `host` and `subfolder` when it is omitted.
+    broker_host: NotRequired[str]
     label: str
+    library_path: NotRequired[str]
+    # Namespace for stored states/cards; defaults to label (or platform)
+    # lowercased when omitted.
+    emulator: NotRequired[str]
+    # Opt in to whole memory-card sync (broker /memory-card). When true, the
+    # legacy per-file /save-file in-game-save path is skipped for this container.
+    memory_card_sync: NotRequired[bool]
+    # Broker dialect. Omitted (or "broker") is the per-emulator mod contract;
+    # "webstation" is the LSIO webstation container's activate/exit contract.
+    protocol: NotRequired[str]
+    # URL prefix the webstation broker is served under, matching its SUBFOLDER.
+    subfolder: NotRequired[str]
 
 
 class Config:

backend/conftest.py

@@ -1,15 +1,16 @@
 import os
 
-# When running under pytest-xdist, give each worker its own database so the
-# autouse `clear_database` fixture in one worker can't wipe rows another worker
-# is mid-test with. This must run before any application module (config /
-# database handlers) is imported, so the engine built at import time binds to
-# the per-worker name. As the rootdir conftest, this file is imported before
-# `tests/conftest.py` (which imports those modules).
+# Tests must never inherit DB_NAME from the ambient environment (e.g. a
+# sourced .env pointing at a real dev/prod database) -- the autouse
+# `clear_database` fixture in tests/conftest.py deletes every row in that
+# database on every test. Always pin to a dedicated test database, made
+# unique per pytest-xdist worker so parallel workers can't wipe rows another
+# worker is mid-test with. This must run before any application module
+# (config / database handlers) is imported, so the engine built at import
+# time binds to the test name. As the rootdir conftest, this file is
+# imported before `tests/conftest.py` (which imports those modules).
 #
 # The Redis cache needs no equivalent handling: under pytest it is an in-process
 # FakeRedis, so each worker process is already isolated.
 _xdist_worker = os.environ.get("PYTEST_XDIST_WORKER")
-if _xdist_worker:
-    _base_db_name = os.environ.get("DB_NAME", "romm_test")
-    os.environ["DB_NAME"] = f"{_base_db_name}_{_xdist_worker}"
+os.environ["DB_NAME"] = f"romm_test_{_xdist_worker}" if _xdist_worker else "romm_test"