feat: set users ID to be autoincrementing

supernote/alembic/versions/68964804740d_prevent_user_id_reuse.py

@@ -0,0 +1,74 @@
+"""Prevent reuse of deleted SQLite user IDs.
+
+Revision ID: 68964804740d
+Revises: d1e2f3a4b5c6
+Create Date: 2026-09-03 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+from sqlalchemy import text
+
+revision: str = "68964804740d"
+down_revision: Union[str, None] = "d1e2f3a4b5c6"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if bind.dialect.name != "sqlite":
+        return
+
+    # Old deletions may have left owner rows behind. Preserve their highest user
+    # ID as part of the sequence floor so none can become visible after upgrade.
+    tables = (
+        "users",
+        "devices",
+        "login_records",
+        "f_user_file",
+        "f_capacity",
+        "f_recycle_file",
+        "f_summary",
+        "f_summary_tag",
+        "t_schedule_task",
+        "t_schedule_task_group",
+    )
+
+    def max_owner_id(table: str) -> int:
+        column = "id" if table == "users" else "user_id"
+        query = text(f'SELECT COALESCE(MAX({column}), 0) FROM "{table}"')
+        return int(bind.execute(query).scalar_one())
+
+    floor = max(max_owner_id(table) for table in tables)
+
+    with op.batch_alter_table(
+        "users", recreate="always", table_kwargs={"sqlite_autoincrement": True}
+    ):
+        pass
+
+    current = int(
+        bind.execute(
+            text(
+                "SELECT COALESCE(MAX(seq), 0) FROM sqlite_sequence WHERE name = 'users'"
+            )
+        ).scalar_one()
+    )
+    sequence = max(floor, current)
+    result = bind.execute(
+        text("UPDATE sqlite_sequence SET seq = :seq WHERE name = 'users'"),
+        {"seq": sequence},
+    )
+    if result.rowcount == 0:
+        bind.execute(
+            text("INSERT INTO sqlite_sequence(name, seq) VALUES ('users', :seq)"),
+            {"seq": sequence},
+        )
+
+
+def downgrade() -> None:
+    if op.get_bind().dialect.name != "sqlite":
+        return
+    with op.batch_alter_table("users", recreate="always"):
+        pass

supernote/server/db/models/user.py

@@ -11,6 +11,8 @@ class UserDO(Base):
     """User database model."""
 
     __tablename__ = "users"
+    # User ids should not be recycled
+    __table_args__ = {"sqlite_autoincrement": True}
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     email: Mapped[str] = mapped_column(String, unique=True, index=True)

tests/server/conftest.py

@@ -189,6 +189,15 @@ async def create_test_user(
         assert result.is_active
 
 
+@pytest.fixture
+async def test_user_id(
+    create_test_user: None,
+    user_service: UserService,
+) -> int:
+    """Return the database ID assigned to the default test user."""
+    return await user_service.get_user_id(TEST_USERNAME)
+
+
 @pytest.fixture(name="auth_headers")
 async def auth_headers_fixture(
     server_config: ServerConfig,

tests/server/db/test_migrations.py

@@ -75,7 +75,7 @@ def test_migration_upgrades_successfully(migrated_db: str) -> None:
         # Check if the 'alembic_version' table exists and has the head revision
         result = conn.execute(text("SELECT version_num FROM alembic_version"))
         version = result.scalar()
-        assert version == "d1e2f3a4b5c6"
+        assert version == "68964804740d"
 
         # Check if a known table exists (e.g. users)
         result = conn.execute(text("SELECT count(*) FROM users"))

tests/server/routes/test_auth.py

@@ -221,17 +221,9 @@ async def test_user_unregister(
     res2 = await user2_web.list_query(directory_id=folder_id)
     user2_files = [f.file_name for f in res2.user_file_vo_list]
 
-    # TODO: https://github.com/allenporter/supernote/pull/240
-    # Currently, SQLite recycles the ROWID of deleted users, and unregister() does not
-    # cascade delete user files from f_user_file. This causes a newly registered user
-    # to obtain the deleted user's ID and inherit access to their files.
-    # When PR #240 adds autoincrementing IDs, update these assertions to:
-    # assert user1_id != user2_id
-    # assert "User1PrivateFolder" not in folder_names
-    # assert len(user2_files) == 0
-    assert user1_id == user2_id
-    assert "User1PrivateFolder" in folder_names
-    assert user2_files == ["secret.txt"]
+    assert user1_id != user2_id
+    assert "User1PrivateFolder" not in folder_names
+    assert len(user2_files) == 0
 
 
 async def test_update_password_and_email(

tests/server/routes/test_extended.py

@@ -52,14 +52,17 @@ def patch_gemini_service(mock_gemini_service: Generator[None]) -> None:
 async def test_extended_search(
     extended_client: ExtendedClient,
     session_manager: DatabaseSessionManager,
+    test_user_id: int,
 ) -> None:
     # 1. Seed some search data
-    user_id = 1
     file_id = 101
     async with session_manager.session() as session:
         session.add(
             UserFileDO(
-                id=file_id, user_id=user_id, file_name="SearchTest.note", directory_id=0
+                id=file_id,
+                user_id=test_user_id,
+                file_name="SearchTest.note",
+                directory_id=0,
             )
         )
         session.add(
@@ -84,14 +87,17 @@ async def test_extended_search_with_mock(
     extended_client: ExtendedClient,
     session_manager: DatabaseSessionManager,
     client: Any,  # TestClient from aiohttp
+    test_user_id: int,
 ) -> None:
     # 1. Seed data
-    user_id = 1
     file_id = 101
     async with session_manager.session() as session:
         session.add(
             UserFileDO(
-                id=file_id, user_id=user_id, file_name="Fox.note", directory_id=0
+                id=file_id,
+                user_id=test_user_id,
+                file_name="Fox.note",
+                directory_id=0,
             )
         )
         session.add(

tests/server/services/test_integrity.py

@@ -92,15 +92,13 @@ async def test_integrity_check(
 
 async def test_integrity_orphans(
     integrity_service: IntegrityService,
-    create_test_user: None,
+    test_user_id: int,
     db_session: AsyncSession,
 ) -> None:
     """Verify integrity check detects orphaned files."""
-    user_id = 1
-
-    # Create the user manually
+    # Create an orphaned VFS entry for the default test user
     file_do = UserFileDO(
-        user_id=user_id,
+        user_id=test_user_id,
         file_name="orphan.txt",
         is_folder="N",
         directory_id=9999,  # Invalid
@@ -109,18 +107,17 @@ async def test_integrity_orphans(
     db_session.add(file_do)
     await db_session.commit()
 
-    report = await integrity_service.verify_user_storage(user_id)
+    report = await integrity_service.verify_user_storage(test_user_id)
     assert report.orphans == 1
 
 
 async def test_integrity_hash_mismatch(
     integrity_service: IntegrityService,
     blob_storage: LocalBlobStorage,
-    create_test_user: None,
+    test_user_id: int,
     db_session: AsyncSession,
 ) -> None:
     """Verify integrity check detects hash mismatch."""
-    user_id = 1
     content = b"content"
 
     # Create a new blob
@@ -130,7 +127,7 @@ async def test_integrity_hash_mismatch(
     # Create VFS entry with WRONG MD5
     bad_md5 = "00000000000000000000000000000000"
     file_do = UserFileDO(
-        user_id=user_id,
+        user_id=test_user_id,
         file_name="bad_hash.txt",
         is_folder="N",
         directory_id=0,
@@ -143,7 +140,7 @@ async def test_integrity_hash_mismatch(
     await db_session.commit()
 
     # Verify
-    report = await integrity_service.verify_user_storage(user_id)
+    report = await integrity_service.verify_user_storage(test_user_id)
     assert report.hash_mismatch == 1
     # 8 Default folders are OK
     assert report.ok == 8
@@ -154,11 +151,10 @@ async def test_integrity_hash_mismatch(
 async def test_integrity_basic(
     integrity_service: IntegrityService,
     blob_storage: LocalBlobStorage,
-    create_test_user: None,
+    test_user_id: int,
 ) -> None:
     """Verify basic integrity check with no issues."""
-    user_id = 1
-    report = await integrity_service.verify_user_storage(user_id)
+    report = await integrity_service.verify_user_storage(test_user_id)
 
     assert report.orphans == 0
     assert report.missing_blob == 0