fix(db): SQLite applies the ON DELETE rules of archives, library files, queue jobs, batches and projects

SQLite runs with foreign keys off, so a delete applied none of the
models' ON DELETE CASCADE / SET NULL rules, and SQLite gives the
highest freed id to the next row: a queue job could print another
archive or library file, a new project showed a deleted project's files,
archives kept a deleted library file's id.

core/sqlite_fk applies the rules of an explicit list of references
(children of print_archives, library_files, print_queue, print_batches
and projects) at every ORM delete: after_delete collects the ids and
after_flush applies each rule once per flush; ORM bulk deletes get them
in do_orm_execute, after flushing pending changes. Printer and spool
references are left out on purpose: their model rules don't match what
the app wants yet. On every SQLite start, queue jobs of a library file
that is gone get the library rule (cancel, keep finished), then the
listed rules are applied to what earlier deletes left behind.

Also:
- printer delete with archives answers 409 and the archive purge skips
  an archive while a queue item for it is printing, as DELETE
  /archives/{id} does
- deleting a user clears their id in the print log
- archive_fts is rebuilt once: archives older than the index were never
  indexed, and updating one failed with "database disk image is
  malformed"
- the external rescan cleans up vanished files in one step
- load_all_models() (cached) replaces three ad-hoc model loaders and
  fixes an order-dependent KeyError in _migrate_pg_columns_to_model
- LibraryFile.active() is typed Select[LibraryFile]; the basedpyright
  baseline drops from 842 to 628

CHANGELOG.md

@@ -162,6 +162,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **SQLite: deleting archives, library files, queue jobs, batches or projects no longer leaves references to them behind** — SQLite doesn't apply the database's "when this row is deleted, clear or delete what points at it" rules, and it gives the highest freed id to the next row created. So a queue job could end up printing a different archive or library file than the one it was made for, a new project could show a deleted project's files and folders, and archives kept the id of a deleted library file (a new file then inherited them in project progress). Every delete of these now applies the rules the way PostgreSQL does, and every start repairs what earlier deletes left behind: queue jobs of a deleted archive are removed, as deleting an archive already does, while those of a library file that is gone are cancelled or kept as finished, as a library delete does. Deleting a user now also clears their id in the print log, so those runs show under "no user". Deleting a printer together with its archives, and the archive purge, now leave an archive alone while one of its queue jobs is printing (the printer delete answers that it has to wait), as deleting a single archive already did. Archives stored before archive search existed are now added to the search index once; until then, editing such an archive could fail with "database disk image is malformed". PostgreSQL installs aren't affected, apart from the printing check.
 - **API: `null` for a field that can't be empty no longer causes a server error** — On most PATCH/PUT routes, `null` for a field leaves it unchanged. About a dozen routes (printers, smart plugs, queue items and bulk edits, archives, filaments, spools, notification providers, GitHub backup, Home Assistant sensors, maintenance, external links) wrote it to the database instead and failed with a server error when the field can't be empty, such as a name or a switch. They now leave it unchanged too, and the rest of the request is applied. `null` for a field that may be empty, such as a printer's location, still clears it. The web interface never sent such values; this affects API clients and scripts.
 - **Webhook routes answer a database problem with 503, not 500** — The `/webhook/...` routes checked API keys with their own copy of the check, which missed earlier fixes. A database error while checking the key, a key too long to hash or a damaged stored key came back as a 500. They now use the same check as every other route: 503 when the key can't be checked right now, 401 when it isn't valid. An expired key now reads "Invalid API key" there too, like everywhere else.
 - **The MQTT relay publishes what the wiki says it does** — Three topics were missing or wrong. `smart_plugs/energy` was never sent: with the relay connected, each enabled plug's power, today's energy and total energy now go out once a minute (without the relay no plug is read for this, and an MQTT plug that stopped reporting is left out). `smart_plugs/on` and `/off` only came from switching a plug by hand; auto-on, auto-off, schedules, the queue powering a printer on and failure detection switching a plug off now publish them too. `maintenance/alert` always said 0 for the current value and the threshold; it now has the hours (or days) since the item was last done, its interval and a `unit` field.

backend/app/api/routes/library.py

@@ -17,7 +17,7 @@ from pathlib import Path
 
 from fastapi import APIRouter, Depends, File, HTTPException, Query, Response, UploadFile
 from fastapi.responses import FileResponse as FastAPIFileResponse
-from sqlalchemy import distinct, func, select, update
+from sqlalchemy import delete, distinct, func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -2146,6 +2146,7 @@ async def scan_external_folder(
     # every scan even though the file is still there. os.path.exists keeps
     # such records; genuinely-deleted files (absent from disk) are still
     # cleaned up. External file_path is the absolute on-disk path.
+    vanished_ids: list[int] = []
     for path_str, db_file in existing_files.items():
         if path_str not in found_paths and not os.path.exists(path_str):
             # Clean up thumbnail if we generated one
@@ -2160,8 +2161,17 @@ async def scan_external_folder(
             # its photos go with it rather than being orphaned under an id
             # nothing points at any more (#3077).
             remove_library_photos_dir(db_file.id)
-            await db.delete(db_file)
-            removed += 1
+            vanished_ids.append(db_file.id)
+    if vanished_ids:
+        # Same clean-up as every other hard delete of library rows, once for
+        # all of them: a share that lost thousands of files must not run it
+        # per file while holding the SQLite write lock.
+        from backend.app.services.library_trash import delete_dependent_variants, release_queue_references
+
+        await delete_dependent_variants(db, vanished_ids)
+        await release_queue_references(db, vanished_ids)
+        await db.execute(delete(LibraryFile).where(LibraryFile.id.in_(vanished_ids)))
+        removed += len(vanished_ids)
 
     # Remove empty subfolders whose directories no longer exist on disk
     # Process deepest-first by sorting on path depth (descending)

backend/app/api/routes/printers.py

@@ -474,6 +474,27 @@ async def delete_printer(
     if not printer:
         raise HTTPException(404, "Printer not found")
 
+    if delete_archives:
+        # Deleting an archive removes its queue items; one that is printing
+        # (a reprint on another printer, say) must not go from under the
+        # dispatcher -- the same 409 as DELETE /archives/{id} (#1734).
+        from backend.app.models.print_queue import PrintQueueItem
+
+        printing = await db.scalar(
+            select(func.count())
+            .select_from(PrintQueueItem)
+            .where(
+                PrintQueueItem.status == "printing",
+                PrintQueueItem.archive_id.in_(select(PrintArchive.id).where(PrintArchive.printer_id == printer_id)),
+            )
+        )
+        if printing:
+            raise HTTPException(
+                409,
+                f"Cannot delete this printer's archives — {printing} queue item(s) using them are "
+                f"currently printing. Stop the print first, or keep the archives.",
+            )
+
     printer_manager.disconnect_printer(printer_id)
 
     if delete_archives:

backend/app/api/routes/users.py

@@ -4,7 +4,7 @@ from typing import Annotated
 import jwt as _jwt
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from fastapi.security import HTTPAuthorizationCredentials
-from sqlalchemy import delete, func, select
+from sqlalchemy import delete, func, select, update as sql_update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -33,6 +33,7 @@ from backend.app.models.library import LibraryFile
 from backend.app.models.long_lived_token import LongLivedToken
 from backend.app.models.oidc_provider import UserOIDCLink
 from backend.app.models.print_batch import PrintBatch
+from backend.app.models.print_log import PrintLogEntry
 from backend.app.models.print_queue import PrintQueueItem
 from backend.app.models.settings import Settings
 from backend.app.models.user import User
@@ -460,7 +461,13 @@ async def delete_user(
         doomed_library_file_ids = list(
             (await db.execute(select(LibraryFile.id).where(LibraryFile.created_by_id == user_id))).scalars().all()
         )
-        # Delete all items created by this user
+        # Delete all items created by this user. The print log stays, without
+        # the archives it pointed at; archives of other users printed from
+        # these library files lose that link (both SET NULL on PostgreSQL).
+        from backend.app.services.library_trash import delete_dependent_variants, release_queue_references
+
+        await delete_dependent_variants(db, doomed_library_file_ids)
+        await release_queue_references(db, doomed_library_file_ids)
         await db.execute(delete(PrintArchive).where(PrintArchive.created_by_id == user_id))
         await db.execute(delete(PrintQueueItem).where(PrintQueueItem.created_by_id == user_id))
         await db.execute(delete(LibraryFile).where(LibraryFile.created_by_id == user_id))
@@ -495,6 +502,10 @@ async def delete_user(
         await db.execute(update(LibraryFile).where(LibraryFile.created_by_id == user_id).values(created_by_id=None))
         await db.execute(update(PrintBatch).where(PrintBatch.created_by_id == user_id).values(created_by_id=None))
 
+    # The print log keeps who started each run as long as the user exists, and
+    # outlives them either way (ON DELETE SET NULL, which SQLite doesn't apply).
+    await db.execute(sql_update(PrintLogEntry).where(PrintLogEntry.created_by_id == user_id).values(created_by_id=None))
+
     # Drop API keys owned by this user. The model declares ON DELETE CASCADE
     # so Postgres handles this automatically, but SQLite ships with FK
     # enforcement off (the project's existing pattern — same reason the

backend/app/core/database.py

@@ -1,4 +1,5 @@
 import asyncio
+import functools
 import logging
 import re
 
@@ -312,6 +313,63 @@ async def get_db() -> AsyncSession:
                 pass
 
 
+@functools.cache
+def load_all_models():
+    """Import every model module and return Base.metadata.
+
+    The models package __init__ doesn't import them all, so code that walks the
+    metadata (every table, every foreign key) needs this to see the whole
+    schema, whatever has happened to be imported before. Cached: it is called
+    per request (schemas/not_null.py) and the models don't change at runtime.
+    """
+    import importlib
+    import pkgutil
+
+    import backend.app.models
+
+    for module in pkgutil.iter_modules(backend.app.models.__path__):
+        importlib.import_module(f"backend.app.models.{module.name}")
+    return Base.metadata
+
+
+async def _repair_sqlite_orphans(conn) -> None:
+    """Repair rows that earlier deletes left pointing at a row that is gone (SQLite only).
+
+    SQLite runs with foreign keys off, so before core/sqlite_fk applied the ON
+    DELETE rules at every delete, none were applied. This applies them now to
+    what was left (sweep_orphans), after the one domain rule that deviates from
+    the model's. PostgreSQL enforces the rules itself.
+    """
+    from sqlalchemy import exists, select
+
+    from backend.app.core.sqlite_fk import sweep_orphans
+    from backend.app.models.library import LibraryFile
+    from backend.app.models.print_queue import PrintQueueItem
+    from backend.app.services.library_trash import release_queue_references
+
+    if not is_sqlite():
+        return
+    # print_queue.library_file_id is ON DELETE CASCADE in the model, but a library
+    # delete keeps the rows (release_queue_references): finished plates count
+    # towards a batch, a printing job stays, a waiting one is cancelled. Rows the
+    # old external rescan and pre-#2819 deletes left behind get the same.
+    vanished = (
+        (
+            await conn.execute(
+                select(PrintQueueItem.library_file_id)
+                .where(PrintQueueItem.library_file_id.is_not(None))
+                .where(~exists().where(LibraryFile.id == PrintQueueItem.library_file_id))
+                .distinct()
+            )
+        )
+        .scalars()
+        .all()
+    )
+    if vanished:
+        await release_queue_references(conn, list(vanished))
+    await sweep_orphans(conn)
+
+
 async def init_db():
     # Import models to register them with SQLAlchemy
     from backend.app.models import (  # noqa: F401
@@ -385,6 +443,10 @@ async def init_db():
         # Run migrations for new columns (SQLite doesn't auto-add columns)
         await run_migrations(conn)
 
+        # SQLite applies no ON DELETE rules (foreign keys off): repair rows that
+        # earlier deletes left pointing at nothing. Not a migration.
+        await _repair_sqlite_orphans(conn)
+
     # Re-encrypt any legacy plaintext OIDC client_secret / TOTP secret rows
     # that exist from before the encryption key was configured.
     # Runs on a fresh AsyncSession (NOT the run_migrations() connection) so it
@@ -2195,6 +2257,23 @@ async def _run_migrations(conn):
         except (OperationalError, ProgrammingError):
             pass  # Already applied
 
+        # One-shot: index the archives that existed before archive_fts did. Its
+        # creation above runs no rebuild, so on such installs those rows were
+        # never indexed, and the AFTER UPDATE trigger's FTS 'delete' on one of
+        # them fails with "database disk image is malformed" -- any UPDATE of an
+        # old archive, including the start-up cleanup in
+        # _migrate_add_missing_set_null_fks, which would then fail every start.
+        # The two backfills below rebuilt only when they had work of their own.
+        fts_flag = "_archive_fts_rebuilt_once"
+        already = (
+            await conn.execute(text('SELECT value FROM settings WHERE "key" = :k'), {"k": fts_flag})
+        ).scalar_one_or_none()
+        if already is None:
+            await conn.execute(text("INSERT INTO archive_fts(archive_fts) VALUES('rebuild')"))
+            await conn.execute(
+                text('INSERT INTO settings ("key", value) VALUES (:k, :v)'), {"k": fts_flag, "v": "true"}
+            )
+
     # Migration: Add auto_off_pending columns to smart_plugs (for restart recovery)
     await _safe_execute(conn, "ALTER TABLE smart_plugs ADD COLUMN auto_off_pending BOOLEAN DEFAULT 0")
     await _safe_execute(conn, "ALTER TABLE smart_plugs ADD COLUMN auto_off_pending_since DATETIME")
@@ -5506,8 +5585,9 @@ async def _migrate_pg_columns_to_model(conn) -> None:
     if is_sqlite():
         return
     compiler = conn.dialect.ddl_compiler(conn.dialect, None)
+    tables = load_all_models().tables
     for table, column in _PG_COLUMNS_SET_FROM_MODEL:
-        model_column = Base.metadata.tables[table].c[column]
+        model_column = tables[table].c[column]
         default = compiler.get_column_default_string(model_column)
         await _safe_execute(conn, f"ALTER TABLE {table} ALTER COLUMN {column} SET DEFAULT {default}")
         if model_column.nullable:
@@ -5626,11 +5706,11 @@ async def _migrate_add_missing_set_null_fks(conn) -> None:
     """Add the foreign keys in _MISSING_SET_NULL_FKS where they are missing.
 
     PostgreSQL only; SQLite can't add a constraint to a table and doesn't
-    enforce foreign keys anyway (PRAGMA foreign_keys is off). A column that
-    already has a foreign key to the parent table, whatever its name, is left
-    alone. Ids left behind by deletes without the constraint are cleared first,
-    which is what ON DELETE SET NULL would have done, and without which the
-    constraint can't be added.
+    enforce foreign keys (PRAGMA foreign_keys is off; init_db sweeps orphans
+    there instead, core/sqlite_fk). A column that already has a foreign key to
+    the parent table, whatever its name, is left alone. Ids left behind by
+    deletes without the constraint are cleared first, which is what ON DELETE
+    SET NULL would have done, and without which the constraint can't be added.
     """
     from sqlalchemy import text
 
@@ -7127,3 +7207,7 @@ async def repair_wallet_ledger_internal(session: AsyncSession):
 
     await session.flush()
     return updated_count
+
+
+# The ON DELETE rules SQLite doesn't apply itself: hooks on every ORM delete.
+from backend.app.core import sqlite_fk  # noqa: E402, F401

backend/app/core/sqlite_fk.py

@@ -0,0 +1,255 @@
+"""Selected ON DELETE rules of the foreign keys, applied by hand on SQLite.
+
+The SQLite database runs with ``PRAGMA foreign_keys`` off: turning it on breaks
+the migrations' table rebuilds (measured: every test errors with "no such table:
+main.print_archives"). So a delete there applies none of the models' ON DELETE
+CASCADE / SET NULL rules. Rows pointing at the deleted row keep its id, and
+SQLite hands the highest freed id to the next row created, which then inherits
+them: a queue item prints another archive, a project shows another project's
+files.
+
+Only the references in APPLIED are handled: those where a dangling id does harm
+and the model's rule is what the app wants. Not every model rule is (deleting a
+printer would make a printer-scoped notification provider global, and delete
+finished queue history), so a reference is added here deliberately, never by
+default.
+
+Hooks apply them at every ORM delete, so no delete path needs to remember:
+
+- ``after_delete`` on every mapped class collects the ids of ``db.delete(obj)``,
+  and ``after_flush`` applies the rules once per flush, one statement per rule;
+- ``do_orm_execute`` applies them before ``db.execute(delete(Model)...)``.
+
+Rule statements go straight to the connection and don't trigger the hooks again.
+``sweep_orphans`` repairs what earlier deletes left behind; init_db runs it at
+every start. Nothing here does anything on PostgreSQL, which enforces the rules
+itself.
+"""
+
+import logging
+from functools import cache
+from typing import Any
+
+from sqlalchemy import Column, Connection, Delete, ForeignKey, Table, delete, event, exists, select, text, update
+from sqlalchemy.orm import Mapper, ORMExecuteState, Session
+
+from backend.app.core.database import Base
+
+logger = logging.getLogger(__name__)
+
+# (child table, child column) whose ON DELETE rule is applied on SQLite.
+APPLIED: frozenset[tuple[str, str]] = frozenset(
+    {
+        # print_archives
+        ("active_print_spoolman", "archive_id"),
+        ("budget_reservations", "print_archive_id"),
+        ("library_folders", "archive_id"),
+        ("pending_uploads", "archived_id"),
+        ("pipeline_runs", "source_archive_id"),
+        ("print_batches", "archive_id"),
+        ("print_log_entries", "archive_id"),
+        ("print_queue", "archive_id"),
+        ("project_bom_items", "archive_id"),
+        ("telegram_pending_verdicts", "archive_id"),
+        ("wallet_transactions", "print_archive_id"),
+        # library_files. print_queue.library_file_id is CASCADE in the model, but
+        # library deletes run release_queue_references first, which keeps the rows
+        # (finished plates count towards a batch); the rule then finds nothing.
+        ("library_file_tags", "file_id"),
+        ("pipeline_runs", "sliced_library_file_id"),
+        ("pipeline_runs", "source_library_file_id"),
+        ("print_archives", "library_file_id"),
+        ("print_batches", "library_file_id"),
+        ("print_queue", "library_file_id"),
+        ("print_queue_variants", "library_file_id"),
+        # print_queue
+        ("pipeline_jobs", "queue_entry_id"),
+        ("print_log_entries", "queue_item_id"),
+        ("print_queue_variants", "queue_item_id"),
+        ("wallet_transactions", "print_queue_id"),
+        # print_batches
+        ("print_batch_plates", "batch_id"),
+        ("print_queue", "batch_id"),
+        # projects
+        ("library_files", "project_id"),
+        ("library_folders", "project_id"),
+        ("pending_uploads", "project_id"),
+        ("print_archives", "project_id"),
+        ("print_batches", "project_id"),
+        ("print_queue", "project_id"),
+        ("project_bom_items", "project_id"),
+    }
+)
+
+_PENDING = "bambuddy_sqlite_fk_pending"
+_TABLES = "bambuddy_sqlite_fk_tables"
+# Keeps any one statement's IN list small.
+_CHUNK = 500
+_MAX_DEPTH = 32
+
+
+def _applies(connection: Connection) -> bool:
+    """Only SQLite: PostgreSQL enforces the rules itself."""
+    return connection.dialect.name == "sqlite"
+
+
+def _existing_tables(connection: Connection) -> set[str]:
+    """Tables the database has (cached on the connection); a model without one has no rows to fix."""
+    tables = connection.info.get(_TABLES)
+    if tables is None:
+        result = connection.execute(text("SELECT name FROM sqlite_master WHERE type = 'table'"))
+        tables = connection.info[_TABLES] = {row[0] for row in result}
+    return tables
+
+
+def _single_primary_key(table: Table) -> Column | None:
+    columns = list(table.primary_key.columns)
+    return columns[0] if len(columns) == 1 else None
+
+
+@cache
+def _applied_foreign_keys() -> tuple[ForeignKey, ...]:
+    """The model foreign keys of APPLIED references to a single-column primary key."""
+    from backend.app.core.database import load_all_models
+
+    found = []
+    for table in load_all_models().sorted_tables:
+        for fk in table.foreign_keys:
+            if (
+                (table.name, fk.parent.name) in APPLIED
+                and fk.ondelete in ("CASCADE", "SET NULL")
+                and isinstance(fk.column.table, Table)
+                and fk.column is _single_primary_key(fk.column.table)
+            ):
+                found.append(fk)
+    return tuple(found)
+
+
+@cache
+def _rules(parent_table: str) -> tuple[tuple[Table, Column, str], ...]:
+    """(child table, child column, ON DELETE action) of the APPLIED references to *parent_table*."""
+    return tuple(
+        (fk.parent.table, fk.parent, str(fk.ondelete))
+        for fk in _applied_foreign_keys()
+        if fk.column.table.name == parent_table and isinstance(fk.parent.table, Table)
+    )
+
+
+def _apply(connection: Connection, parent_table: str, ids: Any, depth: int = 0) -> None:
+    """Apply the rules for rows of *parent_table* whose primary key is in *ids* (a list or a select)."""
+    rules = _rules(parent_table)
+    if not rules or depth > _MAX_DEPTH:
+        return
+    if isinstance(ids, list):
+        for start in range(0, len(ids), _CHUNK):
+            _apply_chunk(connection, rules, ids[start : start + _CHUNK], depth)
+    else:
+        _apply_chunk(connection, rules, ids, depth)
+
+
+def _apply_chunk(connection: Connection, rules: tuple, ids: Any, depth: int) -> None:
+    tables = _existing_tables(connection)
+    for child, column, action in rules:
+        if child.name not in tables:
+            continue
+        if action == "SET NULL":
+            connection.execute(update(child).where(column.in_(ids)).values({column.name: None}))
+            continue
+        child_key = _single_primary_key(child)
+        if child_key is not None:
+            # By depth, not by table: a table that refers to itself recurses too.
+            _apply(connection, child.name, select(child_key).where(column.in_(ids)), depth + 1)
+        connection.execute(delete(child).where(column.in_(ids)))
+
+
+@event.listens_for(Base, "after_delete", propagate=True)
+def _collect_orm_delete(mapper: Mapper, connection: Connection, target: Any) -> None:
+    table = mapper.local_table
+    session = Session.object_session(target)
+    if session is None or not isinstance(table, Table) or not _rules(table.name) or not _applies(connection):
+        return
+    pending = session.info.setdefault(_PENDING, {})
+    pending.setdefault(table.name, []).append(mapper.primary_key_from_instance(target)[0])
+
+
+@event.listens_for(Session, "after_flush")
+def _apply_orm_deletes(session: Session, flush_context: Any) -> None:
+    pending = session.info.pop(_PENDING, None)
+    if not pending:
+        return
+    connection = session.connection()
+    # The parent rows are already gone; with foreign keys off that doesn't stop
+    # the rules from finding their dependents.
+    for table, ids in pending.items():
+        _apply(connection, table, ids)
+
+
+@event.listens_for(Session, "after_soft_rollback")
+def _forget_pending(session: Session, previous_transaction: Any) -> None:
+    # A flush that failed after collecting must not leave ids for the next one.
+    session.info.pop(_PENDING, None)
+
+
+@event.listens_for(Session, "do_orm_execute")
+def _before_bulk_delete(state: ORMExecuteState) -> None:
+    statement = state.statement
+    if not state.is_delete or not isinstance(statement, Delete):
+        return
+    table = statement.table
+    if not isinstance(table, Table) or not _rules(table.name):
+        return
+    key = _single_primary_key(table)
+    if key is None:
+        return
+    connection = state.session.connection()
+    if not _applies(connection):
+        return
+    # The session would flush pending changes only after this hook; flush now so
+    # the rules see them (a queue item just moved off the archive must not go).
+    if state.session.autoflush:
+        state.session.flush()
+    where = statement.whereclause
+    _apply(connection, table.name, select(key).where(where) if where is not None else select(key))
+
+
+async def sweep_orphans(conn: Any) -> None:
+    """Repair APPLIED references to rows that no longer exist (SQLite only).
+
+    A SET NULL reference to a missing row is cleared and a CASCADE row is
+    deleted, which is what the rule would have done at the delete. Repeated
+    until a pass changes nothing, since a deleted row can leave its own
+    dependents behind.
+    """
+
+    def run(sync: Any) -> None:
+        connection = sync.connection() if isinstance(sync, Session) else sync
+        if not _applies(connection):
+            return
+        tables = _existing_tables(connection)
+        rules = [
+            fk for fk in _applied_foreign_keys() if fk.parent.table.name in tables and fk.column.table.name in tables
+        ]
+        for _ in range(10):
+            changed = 0
+            for fk in rules:
+                child, column, referenced = fk.parent.table, fk.parent, fk.column
+                missing = column.is_not(None) & ~exists(
+                    select(1).select_from(referenced.table).where(referenced == column)
+                )
+                if fk.ondelete == "SET NULL":
+                    result = connection.execute(update(child).where(missing).values({column.name: None}))
+                else:
+                    result = connection.execute(delete(child).where(missing))
+                if result.rowcount:
+                    changed += result.rowcount
+                    logger.info(
+                        "%s %d %s row(s) whose %s no longer exists",
+                        "Cleared" if fk.ondelete == "SET NULL" else "Deleted",
+                        result.rowcount,
+                        child.name,
+                        referenced.table.name,
+                    )
+            if not changed:
+                return
+
+    await conn.run_sync(run)

backend/app/models/library.py

@@ -218,7 +218,7 @@ class LibraryFile(Base):
     )
 
     @classmethod
-    def active(cls) -> "Select[tuple[LibraryFile]]":
+    def active(cls) -> "Select[LibraryFile]":
         """Select statement that excludes trashed (soft-deleted) files.
 
         Use this in place of ``select(LibraryFile)`` for any user-facing listing

backend/app/schemas/not_null.py

@@ -39,16 +39,6 @@ class IgnoresNullForRequiredColumns(BaseModel):
 
 
 def _table(name: str):
-    from backend.app.core.database import Base
+    from backend.app.core.database import load_all_models
 
-    if name not in Base.metadata.tables:
-        # Every route imports the model it writes, so this only happens when a
-        # schema is validated on its own (a test, a script).
-        import importlib
-        import pkgutil
-
-        import backend.app.models
-
-        for module in pkgutil.iter_modules(backend.app.models.__path__):
-            importlib.import_module(f"backend.app.models.{module.name}")
-    return Base.metadata.tables[name]
+    return load_all_models().tables[name]

backend/app/services/archive_purge.py

@@ -22,7 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from backend.app.core import database as _database
 from backend.app.models.archive import PrintArchive
 from backend.app.models.settings import Settings
-from backend.app.services.archive import ArchiveService
+from backend.app.services.archive import ArchiveService, _count_related_queue_items
 
 logger = logging.getLogger(__name__)
 
@@ -274,6 +274,12 @@ class ArchivePurgeService:
         for archive_id in ids:
             async with _database.async_session() as delete_db:
                 service = ArchiveService(delete_db)
+                # Both modes remove the archive's queue items; one that is
+                # printing stays with its archive until the next run, as
+                # DELETE /archives/{id} refuses with 409 (#1734).
+                _total, printing = await _count_related_queue_items(delete_db, archive_id)
+                if printing:
+                    continue
                 if purge_stats:
                     # Hard-delete linked PrintLogEntry rows first so their
                     # filament / cost contributions stop counting in /stats.

backend/app/services/library_trash.py

@@ -418,7 +418,9 @@ async def release_queue_references(db: AsyncSession, file_ids: list[int]) -> int
 
     Rows already printing are left in place. One of those is a job on a machine
     right now; the file being deleted is the copy in the library, not the copy
-    the printer is working from. Returns the number of items cancelled.
+    the printer is working from. The files' remaining ON DELETE rules (archives
+    printed from them lose the link) apply at the delete itself, on SQLite
+    through core/sqlite_fk. Returns the number of items cancelled.
     """
     if not file_ids:
         return 0

backend/app/services/print_scheduler.py

@@ -8319,6 +8319,11 @@ class PrintScheduler:
                         # Read while the row is still here; the photos move
                         # below, once the delete has actually committed.
                         consumed_photos = list(library_file.photos or [])
+                        # The delete clears the link on the earlier archives of
+                        # this file (ON DELETE SET NULL; core/sqlite_fk on
+                        # SQLite). This one is loaded, so set it here too, or it
+                        # keeps the stale id in the session.
+                        archive.library_file_id = None
                         await db.delete(library_file)
                         file_path = settings.base_dir / archive.file_path
                         filename = archive.filename

backend/tests/integration/test_sqlite_dangling_references.py

@@ -0,0 +1,312 @@
+"""Deleting users, archives and library files leaves no ids pointing at them on SQLite.
+
+SQLite runs with foreign keys off, so the ON DELETE SET NULL that PostgreSQL
+applies to print_log_entries.created_by_id, print_log_entries.archive_id and
+print_archives.library_file_id never fired: the print log kept a deleted user's
+id (so the entries were missing from "no user"), and archives kept the id of a
+deleted library file, which SQLite hands to the next file created.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from backend.app.models.archive import PrintArchive
+from backend.app.models.library import LibraryFile
+from backend.app.models.print_log import PrintLogEntry
+from backend.app.models.user import User
+
+
+async def _user(db_session, name="leaver") -> User:
+    user = User(username=name, password_hash="x", role="user")
+    db_session.add(user)
+    await db_session.commit()
+    await db_session.refresh(user)
+    return user
+
+
+async def _log_entry(db_session, **fields) -> PrintLogEntry:
+    entry = PrintLogEntry(status="completed", print_name="Benchy", started_at=datetime.now(timezone.utc), **fields)
+    db_session.add(entry)
+    await db_session.commit()
+    await db_session.refresh(entry)
+    return entry
+
+
+async def _reload(db_session, model, row_id):
+    statement = select(model).where(model.id == row_id).execution_options(populate_existing=True)
+    return (await db_session.execute(statement)).scalar_one_or_none()
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_deleting_a_user_clears_their_print_log_entries(async_client: AsyncClient, db_session):
+    user = await _user(db_session)
+    entry = await _log_entry(db_session, created_by_id=user.id)
+
+    response = await async_client.delete(f"/api/v1/users/{user.id}")
+
+    assert response.status_code == 204, response.text
+    assert (await _reload(db_session, PrintLogEntry, entry.id)).created_by_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_deleting_a_user_with_their_items_clears_the_log_entries_of_their_archives(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory
+):
+    user = await _user(db_session)
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id, created_by_id=user.id, with_run=False)
+    entry = await _log_entry(db_session, archive_id=archive.id, created_by_id=user.id)
+
+    response = await async_client.delete(f"/api/v1/users/{user.id}?delete_items=true")
+
+    assert response.status_code == 204, response.text
+    assert await _reload(db_session, PrintArchive, archive.id) is None
+    reloaded = await _reload(db_session, PrintLogEntry, entry.id)
+    assert reloaded.archive_id is None
+    assert reloaded.created_by_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_deleting_a_user_with_their_items_clears_archives_of_their_library_files(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory
+):
+    user = await _user(db_session)
+    library_file = LibraryFile(
+        filename="part.3mf", file_path="library/files/part.3mf", file_type="3mf", file_size=1, created_by_id=user.id
+    )
+    db_session.add(library_file)
+    await db_session.commit()
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id, library_file_id=library_file.id, with_run=False)
+
+    response = await async_client.delete(f"/api/v1/users/{user.id}?delete_items=true")
+
+    assert response.status_code == 204, response.text
+    assert (await _reload(db_session, PrintArchive, archive.id)).library_file_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_deleting_a_printer_with_its_archives_clears_their_log_entries(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory
+):
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id, with_run=False)
+    entry = await _log_entry(db_session, archive_id=archive.id, thumbnail_path="archive/1/thumb.png")
+
+    response = await async_client.delete(f"/api/v1/printers/{printer.id}?delete_archives=true")
+
+    assert response.status_code == 200, response.text
+    assert await _reload(db_session, PrintArchive, archive.id) is None
+    reloaded = await _reload(db_session, PrintLogEntry, entry.id)
+    assert reloaded.archive_id is None
+    # The archive files stay on disk here (unlike a single archive delete), so
+    # the print log keeps its picture.
+    assert reloaded.thumbnail_path == "archive/1/thumb.png"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_deleting_a_library_file_clears_the_archives_printed_from_it(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory
+):
+    library_file = LibraryFile(
+        filename="ext.3mf", file_path="/mnt/share/ext.3mf", file_type="3mf", file_size=1, is_external=True
+    )
+    db_session.add(library_file)
+    await db_session.commit()
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id, library_file_id=library_file.id, with_run=False)
+
+    response = await async_client.delete(f"/api/v1/library/files/{library_file.id}")
+
+    assert response.status_code == 200, response.text
+    assert await _reload(db_session, LibraryFile, library_file.id) is None
+    assert (await _reload(db_session, PrintArchive, archive.id)).library_file_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_any_library_file_delete_clears_the_archives_printed_from_it(
+    db_session, printer_factory, archive_factory
+):
+    """Not only the routes: any delete of the row applies the rule (core/sqlite_fk hooks)."""
+    library_file = LibraryFile(filename="a.3mf", file_path="library/files/a.3mf", file_type="3mf", file_size=1)
+    db_session.add(library_file)
+    await db_session.commit()
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id, library_file_id=library_file.id, with_run=False)
+
+    await db_session.delete(library_file)
+    await db_session.commit()
+
+    assert (await _reload(db_session, PrintArchive, archive.id)).library_file_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_deleting_a_user_with_their_items_releases_other_users_queue_jobs(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory
+):
+    """Another user's job on the deleted user's library file or archive must not
+    keep its id: SQLite gives the highest freed id to the next file or archive,
+    and the job would print that one."""
+    from backend.app.models.print_queue import PrintQueueItem
+
+    owner = await _user(db_session, "owner")
+    other = await _user(db_session, "other")
+    library_file = LibraryFile(
+        filename="part.3mf", file_path="library/files/part.3mf", file_type="3mf", file_size=1, created_by_id=owner.id
+    )
+    db_session.add(library_file)
+    await db_session.commit()
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id, created_by_id=owner.id, with_run=False)
+    on_file = PrintQueueItem(printer_id=printer.id, library_file_id=library_file.id, position=1, created_by_id=other.id)
+    on_archive = PrintQueueItem(printer_id=printer.id, archive_id=archive.id, position=2, created_by_id=other.id)
+    db_session.add_all([on_file, on_archive])
+    await db_session.commit()
+    on_file_id, on_archive_id = on_file.id, on_archive.id
+
+    response = await async_client.delete(f"/api/v1/users/{owner.id}?delete_items=true")
+
+    assert response.status_code == 204, response.text
+    kept = await _reload(db_session, PrintQueueItem, on_file_id)
+    assert kept.library_file_id is None
+    assert kept.status == "cancelled"
+    # ON DELETE CASCADE on print_queue.archive_id, as on PostgreSQL.
+    assert await _reload(db_session, PrintQueueItem, on_archive_id) is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_a_rescan_that_drops_a_vanished_external_file_clears_its_archives(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory, tmp_path, monkeypatch
+):
+    from backend.app.core.config import settings as app_settings
+
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+    monkeypatch.setattr(app_settings, "base_dir", data_dir)
+    monkeypatch.setattr(app_settings, "archive_dir", data_dir / "archive")
+    share = tmp_path / "share"
+    share.mkdir()
+    monkeypatch.setenv("BAMBUDDY_EXTERNAL_ROOTS", str(share))
+    (share / "bracket.stl").write_bytes(b"fakestl")
+
+    folder = await async_client.post(
+        "/api/v1/library/folders/external",
+        json={"name": "Share", "external_path": str(share), "readonly": True, "show_hidden": False},
+    )
+    assert folder.status_code == 200, folder.text
+    folder_id = folder.json()["id"]
+    assert (await async_client.post(f"/api/v1/library/folders/{folder_id}/scan")).json()["added"] == 1
+    file_id = (await async_client.get(f"/api/v1/library/files?folder_id={folder_id}")).json()[0]["id"]
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id, library_file_id=file_id, with_run=False)
+
+    (share / "bracket.stl").unlink()
+    rescan = await async_client.post(f"/api/v1/library/folders/{folder_id}/scan")
+
+    assert rescan.json()["removed"] == 1
+    assert (await _reload(db_session, PrintArchive, archive.id)).library_file_id is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_deleting_a_user_with_their_items_applies_the_rules_to_their_jobs_and_batches(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory
+):
+    """Their queue jobs and batches go with a bulk delete; what points at those
+    goes too (a job's model variants) or loses the link (another user's job in
+    their batch), as on PostgreSQL."""
+    from backend.app.models.print_batch import PrintBatch
+    from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
+
+    owner = await _user(db_session, "owner")
+    other = await _user(db_session, "other")
+    library_file = LibraryFile(filename="v.3mf", file_path="library/files/v.3mf", file_type="3mf", file_size=1)
+    batch = PrintBatch(name="Order", created_by_id=owner.id)
+    db_session.add_all([library_file, batch])
+    await db_session.commit()
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id, with_run=False)
+    owned = PrintQueueItem(printer_id=printer.id, archive_id=archive.id, position=1, created_by_id=owner.id)
+    in_batch = PrintQueueItem(
+        printer_id=printer.id, archive_id=archive.id, position=2, created_by_id=other.id, batch_id=batch.id
+    )
+    db_session.add_all([owned, in_batch])
+    await db_session.commit()
+    variant = PrintQueueVariant(queue_item_id=owned.id, library_file_id=library_file.id, target_model="X1C")
+    db_session.add(variant)
+    await db_session.commit()
+    variant_id, in_batch_id = variant.id, in_batch.id
+
+    response = await async_client.delete(f"/api/v1/users/{owner.id}?delete_items=true")
+
+    assert response.status_code == 204, response.text
+    assert await _reload(db_session, PrintQueueVariant, variant_id) is None
+    assert (await _reload(db_session, PrintQueueItem, in_batch_id)).batch_id is None
+
+
+async def _printing_reprint(db_session, printer_factory, archive_factory):
+    """An archive made on printer A, being reprinted right now on printer B."""
+    from backend.app.models.print_queue import PrintQueueItem
+
+    printer_a = await printer_factory(name="A", serial_number="SA")
+    printer_b = await printer_factory(name="B", serial_number="SB")
+    archive = await archive_factory(printer_a.id, with_run=False)
+    item = PrintQueueItem(printer_id=printer_b.id, archive_id=archive.id, position=1, status="printing")
+    db_session.add(item)
+    await db_session.commit()
+    return printer_a.id, archive, item.id
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_deleting_a_printer_with_its_archives_waits_for_a_running_reprint(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory
+):
+    """Like DELETE /archives/{id}: an archive whose queue item is printing is
+    not deleted from under the dispatcher."""
+    from backend.app.models.print_queue import PrintQueueItem
+
+    printer_a_id, archive, item_id = await _printing_reprint(db_session, printer_factory, archive_factory)
+
+    response = await async_client.delete(f"/api/v1/printers/{printer_a_id}?delete_archives=true")
+
+    assert response.status_code == 409, response.text
+    assert "printing" in response.json()["detail"]
+    assert await _reload(db_session, PrintArchive, archive.id) is not None
+    assert await _reload(db_session, PrintQueueItem, item_id) is not None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.parametrize("purge_stats", [False, True])
+async def test_the_archive_purge_skips_an_archive_that_is_being_printed(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory, purge_stats
+):
+    from datetime import datetime, timedelta, timezone
+
+    from backend.app.models.print_queue import PrintQueueItem
+
+    _, archive, item_id = await _printing_reprint(db_session, printer_factory, archive_factory)
+    archive.created_at = datetime.now(timezone.utc) - timedelta(days=400)
+    await db_session.commit()
+
+    response = await async_client.post(
+        "/api/v1/archives/purge", json={"older_than_days": 365, "purge_stats": purge_stats}
+    )
+
+    assert response.status_code == 200, response.text
+    assert response.json()["deleted"] == 0
+    reloaded = await _reload(db_session, PrintArchive, archive.id)
+    assert reloaded is not None and reloaded.deleted_at is None
+    assert await _reload(db_session, PrintQueueItem, item_id) is not None

backend/tests/unit/services/test_notify_widgets.py

@@ -516,11 +516,13 @@ async def test_restart_drops_rows_of_deleted_provider_before_id_is_reused(setup)
     service, api, factory, clock, states = setup
     await service.tick()
     assert len(await rows(factory)) == 1
-    # Production SQLite does not enforce the cascade, and the in-memory cleanup
-    # job is lost when Bambuddy restarts before it drains.
+    # A provider deleted before the ON DELETE rules were applied on SQLite
+    # (core/sqlite_fk), or by raw SQL that bypasses them, left its rows behind,
+    # and the in-memory cleanup job is lost when Bambuddy restarts before it
+    # drains.
     async with factory() as db:
         await db.execute(text("PRAGMA foreign_keys=OFF"))
-        await db.execute(delete(NotificationProvider).where(NotificationProvider.id == 1))
+        await db.execute(text("DELETE FROM notification_providers WHERE id = 1"))
         await db.commit()
         await db.execute(text("PRAGMA foreign_keys=ON"))
     assert len(await rows(factory)) == 1

backend/tests/unit/test_archive_plate_2603.py

@@ -16,6 +16,7 @@ from backend.app.core.database import Base, run_migrations
 from backend.app.models.archive import PrintArchive
 from backend.app.models.print_queue import PrintQueueItem
 from backend.app.models.printer import Printer
+from backend.app.models.settings import Settings
 
 
 @pytest.fixture
@@ -186,6 +187,9 @@ async def test_run_migrations_adds_column_and_backfills_in_order(force_sqlite_di
             db.add(PrintArchive(id=436, filename="heart 3.gcode.3mf", file_path="x", file_size=1, status="cancelled"))
             await db.flush()
             db.add(PrintQueueItem(id=177, printer_id=printer_id, archive_id=436, status="cancelled", plate_id=22))
+            # An install where the one-shot archive_fts rebuild already ran: the
+            # backfill's own rebuild guard is what must keep this from failing.
+            db.add(Settings(key="_archive_fts_rebuilt_once", value="true"))
             await db.commit()
 
         # Upgrade boot: run_migrations must add print_archives.plate_id, rebuild the

backend/tests/unit/test_sqlite_fk.py

@@ -0,0 +1,209 @@
+"""core/sqlite_fk.py applies the selected ON DELETE rules that SQLite leaves out.
+
+The app's SQLite database runs with PRAGMA foreign_keys off, so a delete left
+every ON DELETE CASCADE / SET NULL undone: rows pointing at the deleted row kept
+its id, and SQLite hands the highest freed id to the next row created.
+"""
+
+import pytest
+from sqlalchemy import delete, event, select, text
+
+from backend.app.core import sqlite_fk
+from backend.app.core.database import load_all_models
+from backend.app.models.archive import PrintArchive
+from backend.app.models.library import LibraryFile
+from backend.app.models.notification import NotificationProvider
+from backend.app.models.print_log import PrintLogEntry
+from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
+from backend.app.models.printer import Printer
+
+
+async def _archive_with_dependents(db):
+    printer = Printer(name="P", serial_number="S1", ip_address="192.0.2.1", access_code="x")
+    library_file = LibraryFile(filename="f.3mf", file_path="l/f.3mf", file_type="3mf", file_size=1)
+    db.add_all([printer, library_file])
+    await db.flush()
+    archive = PrintArchive(
+        printer_id=printer.id, filename="a.3mf", file_path="a/a.3mf", file_size=1, status="completed"
+    )
+    db.add(archive)
+    await db.flush()
+    item = PrintQueueItem(printer_id=printer.id, archive_id=archive.id, position=1)
+    db.add(item)
+    await db.flush()
+    variant = PrintQueueVariant(queue_item_id=item.id, library_file_id=library_file.id, target_model="X1C")
+    run = PrintLogEntry(status="completed", print_name="a", archive_id=archive.id, queue_item_id=item.id)
+    db.add_all([variant, run])
+    await db.commit()
+    return archive.id, item.id, variant.id, run.id
+
+
+async def _exists(db, model, row_id) -> bool:
+    return (await db.execute(select(model.id).where(model.id == row_id))).scalar_one_or_none() is not None
+
+
+async def _run(db, run_id):
+    statement = select(PrintLogEntry).where(PrintLogEntry.id == run_id).execution_options(populate_existing=True)
+    return (await db.execute(statement)).scalar_one()
+
+
+async def _assert_rules_applied(db, item_id, variant_id, run_id):
+    assert not await _exists(db, PrintQueueItem, item_id)  # CASCADE
+    assert not await _exists(db, PrintQueueVariant, variant_id)  # CASCADE of the CASCADE
+    run = await _run(db, run_id)
+    assert run.archive_id is None  # SET NULL
+    assert run.queue_item_id is None  # SET NULL of the cascaded row
+
+
+@pytest.mark.asyncio
+async def test_an_orm_delete_applies_the_rules(db_session):
+    archive_id, item_id, variant_id, run_id = await _archive_with_dependents(db_session)
+
+    await db_session.delete(await db_session.get(PrintArchive, archive_id))
+    await db_session.commit()
+
+    await _assert_rules_applied(db_session, item_id, variant_id, run_id)
+
+
+@pytest.mark.asyncio
+async def test_a_bulk_delete_applies_the_rules(db_session):
+    archive_id, item_id, variant_id, run_id = await _archive_with_dependents(db_session)
+
+    await db_session.execute(delete(PrintArchive).where(PrintArchive.id == archive_id))
+    await db_session.commit()
+
+    await _assert_rules_applied(db_session, item_id, variant_id, run_id)
+
+
+@pytest.mark.asyncio
+async def test_a_bulk_delete_sees_changes_not_flushed_yet(db_session):
+    """A queue item moved off the archive in the same transaction must survive."""
+    archive_id, item_id, _, _ = await _archive_with_dependents(db_session)
+    other = PrintArchive(filename="b.3mf", file_path="a/b.3mf", file_size=1, status="completed")
+    db_session.add(other)
+    await db_session.commit()
+    item = await db_session.get(PrintQueueItem, item_id)
+    item.archive_id = other.id
+
+    await db_session.execute(delete(PrintArchive).where(PrintArchive.id == archive_id))
+    await db_session.commit()
+
+    assert await _exists(db_session, PrintQueueItem, item_id)
+
+
+@pytest.mark.asyncio
+async def test_references_not_on_the_list_are_left_alone(db_session):
+    """Not every model rule is right: deleting a printer must not widen a
+    printer-scoped notification provider (SET NULL means all printers)."""
+    printer = Printer(name="P", serial_number="S2", ip_address="192.0.2.2", access_code="x")
+    db_session.add(printer)
+    await db_session.flush()
+    provider = NotificationProvider(name="n", provider_type="webhook", config="{}", printer_id=printer.id)
+    db_session.add(provider)
+    await db_session.commit()
+    printer_id, provider_id = printer.id, provider.id
+
+    await db_session.execute(delete(Printer).where(Printer.id == printer_id))
+    await db_session.commit()
+
+    scoped = (
+        await db_session.execute(select(NotificationProvider.printer_id).where(NotificationProvider.id == provider_id))
+    ).scalar_one()
+    assert scoped == printer_id
+
+
+@pytest.mark.asyncio
+async def test_many_orm_deletes_in_one_flush_cost_one_statement_per_rule(db_session):
+    """Applied once per flush, not once per row: a folder of thousands of files
+    or a big purge must not run the rules per row while holding the write lock."""
+    archives = [PrintArchive(filename=f"{i}.3mf", file_path=f"a/{i}.3mf", file_size=1, status="x") for i in range(50)]
+    db_session.add_all(archives)
+    await db_session.commit()
+    rule_statements: list[str] = []
+    sync_engine = db_session.bind.sync_engine
+
+    def count(conn, cursor, statement, *args):
+        head = statement.lstrip().upper()
+        if head.startswith(("UPDATE", "DELETE")) and not head.startswith("DELETE FROM PRINT_ARCHIVES"):
+            rule_statements.append(statement)
+
+    event.listen(sync_engine, "before_cursor_execute", count)
+    try:
+        for archive in archives:
+            await db_session.delete(archive)
+        await db_session.commit()
+    finally:
+        event.remove(sync_engine, "before_cursor_execute", count)
+
+    # The rules of print_archives and of the rows they cascade to, once each.
+    assert len(rule_statements) < 30, len(rule_statements)
+
+
+@pytest.mark.asyncio
+async def test_a_rollback_drops_ids_collected_for_a_flush_that_failed(db_session):
+    """Ids collected by a flush that then failed must not be applied by the next one."""
+    await db_session.execute(select(1))  # an open transaction, as during a failed flush
+    db_session.sync_session.info[sqlite_fk._PENDING] = {"print_archives": [1]}
+
+    await db_session.rollback()
+
+    assert sqlite_fk._PENDING not in db_session.sync_session.info
+
+
+@pytest.mark.asyncio
+async def test_nothing_to_do_on_postgresql(db_session, monkeypatch):
+    """PostgreSQL applies the rules itself."""
+    archive_id, item_id, _, run_id = await _archive_with_dependents(db_session)
+    monkeypatch.setattr(sqlite_fk, "_applies", lambda connection: False)
+
+    await db_session.execute(delete(PrintArchive).where(PrintArchive.id == archive_id))
+    await db_session.commit()
+
+    assert await _exists(db_session, PrintQueueItem, item_id)
+    assert (await _run(db_session, run_id)).archive_id == archive_id
+
+
+@pytest.mark.asyncio
+async def test_the_startup_sweep_repairs_rows_whose_parent_is_already_gone(db_session):
+    archive_id, item_id, variant_id, run_id = await _archive_with_dependents(db_session)
+    # Deleted the old way (raw SQL bypasses the hooks), leaving every dependent behind.
+    await db_session.execute(text("DELETE FROM print_archives WHERE id = :id"), {"id": archive_id})
+    await db_session.commit()
+
+    await sqlite_fk.sweep_orphans(await db_session.connection())
+    await db_session.commit()
+
+    await _assert_rules_applied(db_session, item_id, variant_id, run_id)
+
+
+@pytest.mark.asyncio
+async def test_the_sweep_leaves_valid_references_and_unlisted_ones_alone(db_session):
+    archive_id, item_id, variant_id, run_id = await _archive_with_dependents(db_session)
+    provider = NotificationProvider(name="n", provider_type="webhook", config="{}", printer_id=987654)
+    db_session.add(provider)
+    await db_session.commit()
+    provider_id = provider.id
+
+    await sqlite_fk.sweep_orphans(await db_session.connection())
+    await db_session.commit()
+
+    assert await _exists(db_session, PrintQueueItem, item_id)
+    assert await _exists(db_session, PrintQueueVariant, variant_id)
+    assert (await _run(db_session, run_id)).archive_id == archive_id
+    stays = (
+        await db_session.execute(select(NotificationProvider.printer_id).where(NotificationProvider.id == provider_id))
+    ).scalar_one()
+    assert stays == 987654
+
+
+def test_every_listed_reference_is_a_model_foreign_key_with_a_rule():
+    """A typo or a renamed column must not silently drop a rule."""
+    metadata = load_all_models()
+    with_rule = {
+        (table.name, fk.parent.name)
+        for table in metadata.sorted_tables
+        for fk in table.foreign_keys
+        if fk.ondelete in ("CASCADE", "SET NULL")
+    }
+    assert sqlite_fk.APPLIED - with_rule == set()
+    assert len(sqlite_fk._applied_foreign_keys()) == len(sqlite_fk.APPLIED)

backend/tests/unit/test_sqlite_orphan_cleanup.py

@@ -0,0 +1,150 @@
+"""On SQLite, every start repairs rows left behind by earlier deletes.
+
+The delete paths apply the ON DELETE rules now (core/sqlite_fk), but databases
+already hold ids of archives and library files deleted before that (SQLite runs
+with foreign keys off). Every start applies the rules to rows whose referenced
+row is gone, which also catches a path that still leaves one behind. The rules
+themselves are tested in test_sqlite_fk.py.
+"""
+
+import importlib
+import pkgutil
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from backend.app.core.database import init_db, run_migrations
+from backend.app.core.sqlite_fk import sweep_orphans
+
+
+@pytest.fixture(autouse=True)
+def force_sqlite_dialect(monkeypatch):
+    from backend.app.core import database as database_module, db_dialect
+
+    monkeypatch.setattr(db_dialect, "is_sqlite", lambda: True)
+    monkeypatch.setattr(db_dialect, "is_postgres", lambda: False)
+    monkeypatch.setattr(database_module, "is_sqlite", lambda: True)
+
+
+@pytest.fixture
+async def engine(tmp_path):
+    import backend.app.models
+    from backend.app.core.database import Base
+
+    for module in pkgutil.iter_modules(backend.app.models.__path__):
+        importlib.import_module(f"backend.app.models.{module.name}")
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/orphans.db")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    yield engine
+    await engine.dispose()
+
+
+async def test_orphaned_ids_are_cleared_and_valid_ones_kept(engine):
+    async with engine.begin() as conn:
+        await conn.execute(
+            text(
+                "INSERT INTO users (id, username, role, is_active, auth_source) VALUES (1, 'kept', 'user', 1, 'local')"
+            )
+        )
+        await conn.execute(
+            text(
+                "INSERT INTO library_files (id, filename, file_path, file_type, file_size, print_count, is_external,"
+                " variant_position) VALUES (1, 'kept.3mf', 'l/kept.3mf', '3mf', 1, 0, 0, 0)"
+            )
+        )
+        await conn.execute(
+            text(
+                "INSERT INTO print_archives (id, filename, file_path, file_size, status, quantity, is_favorite,"
+                " library_file_id, bambuddy_forced_timelapse, wallet_charge_skipped) VALUES"
+                " (1, 'a.3mf', 'a/a.3mf', 1, 'completed', 1, 0, 1, 0, 0),"
+                " (2, 'b.3mf', 'a/b.3mf', 1, 'completed', 1, 0, 999, 0, 0)"
+            )
+        )
+        await conn.execute(
+            text(
+                "INSERT INTO print_log_entries (id, status, print_name, archive_id, created_by_id) VALUES"
+                " (1, 'completed', 'a', 1, 1), (2, 'completed', 'zzz-no-archive-of-this-name', 999, NULL)"
+            )
+        )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+        await sweep_orphans(conn)
+
+    async with engine.connect() as conn:
+        archives = dict((await conn.execute(text("SELECT id, library_file_id FROM print_archives"))).all())
+        entries = {
+            row[0]: row[1:]
+            for row in (await conn.execute(text("SELECT id, archive_id, created_by_id FROM print_log_entries"))).all()
+        }
+
+    assert archives == {1: 1, 2: None}
+    assert entries == {1: (1, 1), 2: (None, None)}
+
+
+async def test_archives_older_than_the_search_index_are_indexed_once(engine):
+    """An archive stored before archive_fts existed was never indexed, and
+    updating it failed with "database disk image is malformed"."""
+    async with engine.begin() as conn:
+        await conn.execute(
+            text(
+                "INSERT INTO print_archives (id, filename, file_path, file_size, status, quantity, is_favorite,"
+                " print_name, bambuddy_forced_timelapse, wallet_charge_skipped) VALUES"
+                " (7, 'old.3mf', 'a/old.3mf', 1, 'completed', 1, 0, 'Gearbox', 0, 0)"
+            )
+        )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+    async with engine.begin() as conn:
+        await conn.execute(text("UPDATE print_archives SET notes = 'kept' WHERE id = 7"))
+        found = (await conn.execute(text("SELECT rowid FROM archive_fts WHERE archive_fts MATCH 'Gearbox'"))).all()
+        flags = (
+            await conn.execute(text("SELECT COUNT(*) FROM settings WHERE key = '_archive_fts_rebuilt_once'"))
+        ).scalar()
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    assert found == [(7,)]
+    assert flags == 1
+
+
+def test_init_db_sweeps_after_the_migrations():
+    """The sweep is a start-up repair, not a migration: init_db runs it."""
+    import inspect
+
+    source = inspect.getsource(init_db)
+    assert source.index("_repair_sqlite_orphans(") > source.index("run_migrations(")
+
+
+async def test_queue_jobs_of_a_vanished_library_file_follow_the_library_rule_not_the_cascade(engine):
+    """print_queue.library_file_id is ON DELETE CASCADE in the model, but a library
+    delete deliberately keeps the rows (release_queue_references): finished plates
+    are what a batch counts its progress from, a printing job stays, a waiting one
+    is cancelled. Rows orphaned by older versions are repaired the same way."""
+    from sqlalchemy.ext.asyncio import async_sessionmaker
+
+    from backend.app.core.database import _repair_sqlite_orphans
+    from backend.app.models.print_queue import PrintQueueItem
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+    async with async_sessionmaker(engine)() as db:
+        db.add_all(
+            [
+                PrintQueueItem(id=1, library_file_id=999, position=1, status="completed"),
+                PrintQueueItem(id=2, library_file_id=999, position=2, status="printing"),
+                PrintQueueItem(id=3, library_file_id=999, position=3, status="pending"),
+            ]
+        )
+        await db.commit()
+
+    async with engine.begin() as conn:
+        await _repair_sqlite_orphans(conn)
+
+    async with engine.connect() as conn:
+        rows = (await conn.execute(text("SELECT id, status, library_file_id FROM print_queue ORDER BY id"))).all()
+
+    assert rows == [(1, "completed", None), (2, "printing", None), (3, "cancelled", None)]