synctv/synctv-core/tests/transaction_isolation_tests.rs

//! Transaction isolation tests for concurrent room member updates
//!
//! These tests verify that concurrent member updates maintain data integrity
//! and isolation guarantees when multiple transactions attempt to modify the same
//! room members simultaneously.
//!
//! Requires Docker for testcontainers.

use sqlx::PgPool;
use std::sync::Arc;
use synctv_core::{
    models::{
        Room, RoomId, RoomMember, RoomRole, RoomStatus, SignupMethod, User, UserId, UserStatus,
    },
    repository::{RoomMemberRepository, RoomRepository},
};
use synctv_core_testing::{create_test_pool, ok, some};
use tokio::sync::Barrier;

/// Default `PostgreSQL` version for test containers
/// Helper to create a test database pool with proper schema
/// Create a test user in the database (required for FK constraints)
async fn create_test_user(pool: &PgPool, user_id: &UserId) {
    let username = format!("test_user_{user_id}");
    let user = User {
        id: *user_id,
        username,
        signup_method: SignupMethod::Email,
        role: synctv_core::models::UserRole::User,
        avatar_file_reference_id: None,
        status: UserStatus::Active,
        created_at: chrono::Utc::now(),
        updated_at: chrono::Utc::now(),
        version: 0,
        deleted_at: None,
        is_banned: false,
        banned_at: None,
        banned_by: None,
        banned_reason: None,
    };
    ok(
        sqlx::query!(
            r"
        INSERT INTO users (
            id, username, signup_method, role,
            created_at, updated_at
        )
        VALUES ($1, $2, $3, $4, $5, $6)
        ",
            user.id.as_i64(),
            &user.username,
            i16::from(user.signup_method),
            i16::from(user.role),
            user.created_at,
            user.updated_at
        )
        .execute(pool)
        .await,
        "test user should be inserted",
    );

    ok(
        sqlx::query!(
            r"
        INSERT INTO auth_email_identities (
            user_id, email, created_at, updated_at
        )
        VALUES ($1, $2, $3, $4)
        ",
            user.id.as_i64(),
            &format!("{user_id}@test.com"),
            user.created_at,
            user.updated_at
        )
        .execute(pool)
        .await,
        "test user email identity should be inserted",
    );

    ok(
        sqlx::query!(
            r"
        INSERT INTO auth_password_credentials (
            user_id, opaque_record, opaque_credential_identifier, opaque_ciphersuite,
            opaque_server_setup_version,
            changed_at, version, created_at, updated_at
        )
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
        ",
            user.id.as_i64(),
            b"test-opaque-record".as_slice(),
            b"test-opaque-id".as_slice(),
            "opaque-ristretto255-sha512-argon2id",
            1_i32,
            user.created_at,
            0_i32,
            user.created_at,
            user.updated_at
        )
        .execute(pool)
        .await,
        "test user password credential should be inserted",
    );
}

fn make_member(room_id: RoomId, user_id: UserId) -> RoomMember {
    RoomMember {
        room_id,
        user_id,
        role: RoomRole::Member,
        added_permissions: 0,
        removed_permissions: 0,
        admin_added_permissions: 0,
        admin_removed_permissions: 0,
        remark_name: String::new(),
        display_tag: String::new(),
        joined_at: chrono::Utc::now(),
        version: 0,
    }
}

#[tokio::test]
#[ignore = "Requires Docker"]
async fn test_concurrent_member_role_updates_isolated() {
    let (_container, pool) = create_test_pool().await;

    let room_repo = RoomRepository::new(pool.clone());
    let member_repo = RoomMemberRepository::new(pool.clone());

    let creator_id = UserId::new();
    create_test_user(&pool, &creator_id).await;
    let room = Room {
        id: RoomId::new(),
        name: "Test Room".to_string(),
        description: "Test".to_string(),
        cover_file_reference_id: None,
        category: None,
        labels: Vec::new(),
        created_by: creator_id,
        status: RoomStatus::Active,
        is_banned: false,
        is_public: true,
        closed_at: None,
        created_at: chrono::Utc::now(),
        updated_at: chrono::Utc::now(),
        deleted_at: None,
        version: 0,
        last_activity_at: chrono::Utc::now(),
    };
    let room = ok(room_repo.create(&room).await, "room should be created");

    let user1 = UserId::new();
    let user2 = UserId::new();
    create_test_user(&pool, &user1).await;
    create_test_user(&pool, &user2).await;

    let member1 = make_member(room.id, user1);
    let member2 = make_member(room.id, user2);

    ok(
        member_repo.add(&member1).await,
        "first member should be added",
    );
    ok(
        member_repo.add(&member2).await,
        "second member should be added",
    );

    // Concurrent role updates using added_permissions column
    let barrier = Arc::new(Barrier::new(2));
    let pool1 = pool.clone();
    let pool2 = pool.clone();
    let room_id = room.id;
    let user1_clone = user1;
    let user2_clone = user2;

    let barrier1 = barrier.clone();
    let handle1 = tokio::spawn(async move {
        barrier1.wait().await;

        let mut tx = pool1.begin().await?;

        sqlx::query!(
            "UPDATE room_members SET added_permissions = $1
             WHERE room_id = $2 AND user_id = $3",
            0xFF_i64,
            room_id.as_i64(),
            user1_clone.as_i64()
        )
        .execute(&mut *tx)
        .await?;

        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

        tx.commit().await?;
        Ok::<(), sqlx::Error>(())
    });

    let barrier2 = barrier.clone();
    let room_id2 = room.id;
    let handle2 = tokio::spawn(async move {
        barrier2.wait().await;

        let mut tx = pool2.begin().await?;

        sqlx::query!(
            "UPDATE room_members SET added_permissions = $1
             WHERE room_id = $2 AND user_id = $3",
            0xFF_i64,
            room_id2.as_i64(),
            user2_clone.as_i64()
        )
        .execute(&mut *tx)
        .await?;

        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

        tx.commit().await?;
        Ok::<(), sqlx::Error>(())
    });

    let update1 = ok(handle1.await, "first update task should complete");
    ok(update1, "first update transaction should finish");
    let update2 = ok(handle2.await, "second update task should complete");
    ok(update2, "second update transaction should finish");

    // Verify both updates succeeded
    let member1_updated = some(
        ok(
            member_repo.get(&room.id, &user1).await,
            "first member should be fetched",
        ),
        "first member should exist",
    );
    let member2_updated = some(
        ok(
            member_repo.get(&room.id, &user2).await,
            "second member should be fetched",
        ),
        "second member should exist",
    );

    assert_eq!(member1_updated.added_permissions, 0xFF);
    assert_eq!(member2_updated.added_permissions, 0xFF);
}

#[tokio::test]
#[ignore = "Requires Docker"]
async fn test_serializable_isolation_prevents_phantom_reads() {
    let (_container, pool) = create_test_pool().await;

    let room_repo = RoomRepository::new(pool.clone());

    let creator_id = UserId::new();
    create_test_user(&pool, &creator_id).await;
    let room = Room {
        id: RoomId::new(),
        name: "Test Room".to_string(),
        description: "Test".to_string(),
        cover_file_reference_id: None,
        category: None,
        labels: Vec::new(),
        created_by: creator_id,
        status: RoomStatus::Active,
        is_banned: false,
        is_public: true,
        closed_at: None,
        created_at: chrono::Utc::now(),
        updated_at: chrono::Utc::now(),
        deleted_at: None,
        version: 0,
        last_activity_at: chrono::Utc::now(),
    };
    let room = ok(room_repo.create(&room).await, "room should be created");

    // Transaction 1: Count members, then re-count
    let pool1 = pool.clone();
    let room_id1 = room.id;
    let handle1 = tokio::spawn(async move {
        let mut tx = pool1.begin().await?;

        let count: i64 = sqlx::query_scalar!(
            r#"SELECT COUNT(*) AS "count!" FROM room_members WHERE room_id = $1"#,
            room_id1.as_i64()
        )
        .fetch_one(&mut *tx)
        .await?;

        assert_eq!(count, 0);

        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

        let count2: i64 = sqlx::query_scalar!(
            r#"SELECT COUNT(*) AS "count!" FROM room_members WHERE room_id = $1"#,
            room_id1.as_i64()
        )
        .fetch_one(&mut *tx)
        .await?;

        tx.commit().await?;

        Ok::<i64, sqlx::Error>(count2)
    });

    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

    let pool2 = pool.clone();
    let room_id2 = room.id;
    let user_id = UserId::new();
    create_test_user(&pool, &user_id).await;
    let handle2 = tokio::spawn(async move {
        let member = make_member(room_id2, user_id);
        let member_repo = RoomMemberRepository::new(pool2);
        member_repo.add(&member).await
    });

    let insert_result = ok(handle2.await, "member insert task should complete");
    ok(insert_result, "member should be added");
    let count_result = ok(handle1.await, "count task should complete");
    let final_count = ok(count_result, "count transaction should finish");

    // Due to READ COMMITTED isolation (default), the re-count may see the new member.
    // This is expected behavior for READ COMMITTED.
    assert!(final_count >= 0);
}