// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

//! Store tests. The portable behaviour runs against SQLite both in-memory and
//! file-backed; the same round-trips run against Postgres when the
//! `postgres-tests` feature is enabled and `FABRICA_TEST_POSTGRES_URL` is set.

#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

use model::KeyKind;
use tempfile::TempDir;

use super::{Backend, NewKey, NewRepo, NewToken, NewUser, Store, StoreError};

/// A migrated store plus any temp directory that must outlive it.
struct Fixture {
    store: Store,
    _dir: Option<TempDir>,
}

/// A migrated in-memory SQLite store. A single connection keeps the database
/// alive for the life of the pool.
async fn memory_store() -> Fixture {
    let store = Store::connect("sqlite::memory:", 1).await.unwrap();
    store.migrate().await.unwrap();
    Fixture { store, _dir: None }
}

/// A migrated file-backed SQLite store in a fresh temp directory.
async fn file_store() -> Fixture {
    let dir = tempfile::tempdir().unwrap();
    let url = format!("sqlite://{}", dir.path().join("fabrica.db").display());
    let store = Store::connect(&url, 5).await.unwrap();
    store.migrate().await.unwrap();
    Fixture {
        store,
        _dir: Some(dir),
    }
}

/// Both SQLite flavours the portable tests must pass against.
async fn sqlite_fixtures() -> Vec<Fixture> {
    vec![memory_store().await, file_store().await]
}

fn sample_user(username: &str, email: &str) -> NewUser {
    NewUser {
        username: username.to_string(),
        email: email.to_string(),
        display_name: Some("Sample".to_string()),
        password_hash: None,
        is_admin: false,
        must_change_password: false,
    }
}

#[tokio::test]
async fn migrations_are_idempotent() {
    for fx in sqlite_fixtures().await {
        // A second run over the already-migrated database is a no-op.
        fx.store.migrate().await.unwrap();
    }
}

#[tokio::test]
async fn create_and_fetch_user() {
    for fx in sqlite_fixtures().await {
        let created = fx
            .store
            .create_user(sample_user("hanna", "hanna@example.com"))
            .await
            .unwrap();
        assert_eq!(created.id.len(), 26, "id should be a 26-char ULID");
        assert_eq!(created.created_at, created.updated_at);

        let by_id = fx.store.user_by_id(&created.id).await.unwrap().unwrap();
        assert_eq!(by_id, created);

        // Username lookup is case-insensitive.
        let by_name = fx.store.user_by_username("HANNA").await.unwrap().unwrap();
        assert_eq!(by_name, created);

        assert!(fx.store.user_by_id("nope").await.unwrap().is_none());
        assert!(fx.store.user_by_username("ghost").await.unwrap().is_none());
    }
}

#[tokio::test]
async fn update_profile_and_avatar_round_trip() {
    for fx in sqlite_fixtures().await {
        let user = fx
            .store
            .create_user(sample_user("prof", "prof@example.com"))
            .await
            .unwrap();

        let ok = fx
            .store
            .update_profile(
                &user.id,
                crate::ProfileUpdate {
                    display_name: Some("Professor".to_string()),
                    pronouns: Some("they/them".to_string()),
                    bio: Some("Hello there.".to_string()),
                    location: Some("Everywhere".to_string()),
                    links: vec![
                        "https://example.com".to_string(),
                        "https://github.com/prof".to_string(),
                    ],
                },
            )
            .await
            .unwrap();
        assert!(ok);

        let got = fx.store.user_by_id(&user.id).await.unwrap().unwrap();
        assert_eq!(got.pronouns.as_deref(), Some("they/them"));
        assert_eq!(got.bio.as_deref(), Some("Hello there."));
        assert_eq!(got.links.len(), 2, "two links stored");
        assert_eq!(got.links[1], "https://github.com/prof");
        assert!(got.avatar_mime.is_none(), "no avatar yet");

        // Setting then clearing the avatar content type round-trips.
        assert!(
            fx.store
                .set_avatar_mime(&user.id, Some("image/png"))
                .await
                .unwrap()
        );
        let got = fx.store.user_by_id(&user.id).await.unwrap().unwrap();
        assert_eq!(got.avatar_mime.as_deref(), Some("image/png"));

        assert!(fx.store.set_avatar_mime(&user.id, None).await.unwrap());
        let got = fx.store.user_by_id(&user.id).await.unwrap().unwrap();
        assert!(got.avatar_mime.is_none(), "avatar cleared");

        // A missing user reports no update.
        assert!(
            !fx.store
                .update_profile("nope", crate::ProfileUpdate::default())
                .await
                .unwrap()
        );
    }
}

#[tokio::test]
async fn labels_create_list_delete() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("lo", "lo@example.com"))
            .await
            .unwrap();
        let repo = fx
            .store
            .create_repo(NewRepo {
                owner_id: owner.id,
                group_id: None,
                name: "r".to_string(),
                path: "r".to_string(),
                description: None,
                visibility: model::Visibility::Private,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap();
        // New repos default to issues/PRs enabled.
        assert!(repo.issues_enabled && repo.pulls_enabled);

        let label = fx
            .store
            .create_label(&repo.id, "type: backend", "#ff0000", Some("server code"))
            .await
            .unwrap();
        assert_eq!(label.scope(), Some("type"));

        // Duplicate name (case-insensitive) conflicts.
        let dup = fx
            .store
            .create_label(&repo.id, "Type: Backend", "#00ff00", None)
            .await;
        assert!(matches!(dup, Err(StoreError::Conflict { .. })));

        let list = fx.store.list_labels(&repo.id).await.unwrap();
        assert_eq!(list.len(), 1);

        assert!(fx.store.delete_label(&label.id).await.unwrap());
        assert!(fx.store.list_labels(&repo.id).await.unwrap().is_empty());

        // Feature toggles round-trip.
        assert!(
            fx.store
                .set_feature_enabled(&repo.id, "issues", false)
                .await
                .unwrap()
        );
        let repo = fx.store.repo_by_id(&repo.id).await.unwrap().unwrap();
        assert!(!repo.issues_enabled);
    }
}

#[tokio::test]
async fn collaborators_add_update_list_remove() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("owner", "owner@example.com"))
            .await
            .unwrap();
        let friend = fx
            .store
            .create_user(sample_user("friend", "friend@example.com"))
            .await
            .unwrap();
        let repo = fx
            .store
            .create_repo(NewRepo {
                owner_id: owner.id.clone(),
                group_id: None,
                name: "proj".to_string(),
                path: "proj".to_string(),
                description: None,
                visibility: model::Visibility::Private,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap();

        fx.store
            .set_collaborator(&repo.id, &friend.id, "read")
            .await
            .unwrap();
        let list = fx.store.list_collaborators(&repo.id).await.unwrap();
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].username, "friend");
        assert_eq!(list[0].permission, "read");

        // Upsert changes the permission in place.
        fx.store
            .set_collaborator(&repo.id, &friend.id, "write")
            .await
            .unwrap();
        let list = fx.store.list_collaborators(&repo.id).await.unwrap();
        assert_eq!(list.len(), 1, "upsert, not a second row");
        assert_eq!(list[0].permission, "write");
        assert_eq!(
            fx.store
                .collaborator_permission(&repo.id, &friend.id)
                .await
                .unwrap()
                .as_deref(),
            Some("write")
        );

        assert!(
            fx.store
                .remove_collaborator(&repo.id, &friend.id)
                .await
                .unwrap()
        );
        assert!(
            fx.store
                .list_collaborators(&repo.id)
                .await
                .unwrap()
                .is_empty()
        );
        assert!(
            !fx.store
                .remove_collaborator(&repo.id, &friend.id)
                .await
                .unwrap()
        );
    }
}

#[tokio::test]
async fn duplicate_username_conflicts_case_insensitively() {
    for fx in sqlite_fixtures().await {
        fx.store
            .create_user(sample_user("dup", "a@example.com"))
            .await
            .unwrap();
        let err = fx
            .store
            .create_user(sample_user("DUP", "b@example.com"))
            .await
            .unwrap_err();
        assert!(
            matches!(
                err,
                StoreError::Conflict {
                    entity: "user",
                    field: "username"
                }
            ),
            "got {err:?}"
        );
    }
}

#[tokio::test]
async fn duplicate_email_conflicts_case_insensitively() {
    for fx in sqlite_fixtures().await {
        fx.store
            .create_user(sample_user("one", "same@example.com"))
            .await
            .unwrap();
        let err = fx
            .store
            .create_user(sample_user("two", "SAME@example.com"))
            .await
            .unwrap_err();
        assert!(
            matches!(
                err,
                StoreError::Conflict {
                    entity: "user",
                    field: "email"
                }
            ),
            "got {err:?}"
        );
    }
}

#[tokio::test]
async fn invalid_username_is_rejected_before_insert() {
    for fx in sqlite_fixtures().await {
        let err = fx
            .store
            .create_user(sample_user("bad name", "x@example.com"))
            .await
            .unwrap_err();
        assert!(matches!(err, StoreError::Name(_)), "got {err:?}");
    }
}

#[tokio::test]
async fn booleans_round_trip() {
    for fx in sqlite_fixtures().await {
        let boss = fx
            .store
            .create_user(NewUser {
                is_admin: true,
                must_change_password: true,
                ..sample_user("boss", "boss@example.com")
            })
            .await
            .unwrap();
        assert!(boss.is_admin);
        assert!(boss.must_change_password);

        let fetched = fx.store.user_by_id(&boss.id).await.unwrap().unwrap();
        assert!(fetched.is_admin, "is_admin must survive the round-trip");
        assert!(fetched.must_change_password);

        // And the default-false path.
        let plain = fx.store.user_by_username("boss").await.unwrap();
        assert!(plain.is_some());
        let normal = fx
            .store
            .create_user(sample_user("normal", "normal@example.com"))
            .await
            .unwrap();
        let normal = fx.store.user_by_id(&normal.id).await.unwrap().unwrap();
        assert!(!normal.is_admin);
        assert!(!normal.must_change_password);
    }
}

/// Insert a bare session row for `user_id` so the session-deletion side effects
/// of `set_password`/`set_disabled` can be observed. Returns nothing; the count
/// is read back with [`session_count`].
async fn insert_session(store: &Store, user_id: &str) {
    sqlx::query(
        "INSERT INTO sessions (id, user_id, created_at, last_seen_at, expires_at) \
         VALUES ($1, $2, 0, 0, 0)",
    )
    .bind(super::new_id())
    .bind(user_id)
    .execute(store.pool())
    .await
    .unwrap();
}

async fn session_count(store: &Store, user_id: &str) -> i64 {
    sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM sessions WHERE user_id = $1")
        .bind(user_id)
        .fetch_one(store.pool())
        .await
        .unwrap()
}

#[tokio::test]
async fn set_password_sets_clears_and_kills_sessions() {
    for fx in sqlite_fixtures().await {
        let user = fx
            .store
            .create_user(sample_user("pw", "pw@example.com"))
            .await
            .unwrap();
        assert!(user.password_hash.is_none());
        insert_session(&fx.store, &user.id).await;
        assert_eq!(session_count(&fx.store, &user.id).await, 1);

        // Setting a hash records it, clears must_change_password, and wipes
        // sessions.
        assert!(
            fx.store
                .set_password(&user.id, Some("$argon2id$fake"))
                .await
                .unwrap()
        );
        let after = fx.store.user_by_id(&user.id).await.unwrap().unwrap();
        assert_eq!(after.password_hash.as_deref(), Some("$argon2id$fake"));
        assert!(!after.must_change_password);
        assert!(after.updated_at >= user.updated_at);
        assert_eq!(session_count(&fx.store, &user.id).await, 0);

        // Clearing it returns to the invite-pending state.
        assert!(fx.store.set_password(&user.id, None).await.unwrap());
        let cleared = fx.store.user_by_id(&user.id).await.unwrap().unwrap();
        assert!(cleared.password_hash.is_none());

        // A missing user reports no update.
        assert!(!fx.store.set_password("ghost", Some("x")).await.unwrap());
    }
}

#[tokio::test]
async fn set_disabled_toggles_and_kills_sessions() {
    for fx in sqlite_fixtures().await {
        let user = fx
            .store
            .create_user(sample_user("dis", "dis@example.com"))
            .await
            .unwrap();
        insert_session(&fx.store, &user.id).await;

        assert!(fx.store.set_disabled(&user.id, true).await.unwrap());
        let disabled = fx.store.user_by_id(&user.id).await.unwrap().unwrap();
        assert!(disabled.disabled_at.is_some());
        assert_eq!(session_count(&fx.store, &user.id).await, 0);

        assert!(fx.store.set_disabled(&user.id, false).await.unwrap());
        let enabled = fx.store.user_by_id(&user.id).await.unwrap().unwrap();
        assert!(enabled.disabled_at.is_none());

        assert!(!fx.store.set_disabled("ghost", true).await.unwrap());
    }
}

#[tokio::test]
async fn list_users_is_sorted_case_insensitively() {
    for fx in sqlite_fixtures().await {
        for name in ["Charlie", "alice", "Bob"] {
            fx.store
                .create_user(sample_user(name, &format!("{name}@example.com")))
                .await
                .unwrap();
        }
        let names: Vec<String> = fx
            .store
            .list_users()
            .await
            .unwrap()
            .into_iter()
            .map(|u| u.username)
            .collect();
        assert_eq!(names, ["alice", "Bob", "Charlie"]);
    }
}

#[tokio::test]
async fn list_users_paged_slices_by_limit_and_offset() {
    for fx in sqlite_fixtures().await {
        for name in ["alice", "Bob", "Charlie", "dave"] {
            fx.store
                .create_user(sample_user(name, &format!("{name}@example.com")))
                .await
                .unwrap();
        }
        let page1: Vec<String> = fx
            .store
            .list_users_paged(2, 0)
            .await
            .unwrap()
            .into_iter()
            .map(|u| u.username)
            .collect();
        assert_eq!(page1, ["alice", "Bob"]);
        let page2: Vec<String> = fx
            .store
            .list_users_paged(2, 2)
            .await
            .unwrap()
            .into_iter()
            .map(|u| u.username)
            .collect();
        assert_eq!(page2, ["Charlie", "dave"]);
        // Past the end yields nothing.
        assert!(fx.store.list_users_paged(2, 4).await.unwrap().is_empty());
    }
}

#[tokio::test]
async fn delete_user_cascades_to_repos() {
    for fx in sqlite_fixtures().await {
        let user = fx
            .store
            .create_user(sample_user("gone", "gone@example.com"))
            .await
            .unwrap();
        fx.store
            .create_repo(NewRepo {
                owner_id: user.id.clone(),
                group_id: None,
                name: "r".to_string(),
                path: "r".to_string(),
                description: None,
                visibility: model::Visibility::Private,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap();

        assert!(fx.store.delete_user(&user.id).await.unwrap());
        assert!(fx.store.user_by_id(&user.id).await.unwrap().is_none());
        assert!(
            fx.store.repos_by_owner(&user.id).await.unwrap().is_empty(),
            "the repo should cascade away with its owner"
        );
        assert!(!fx.store.delete_user(&user.id).await.unwrap());
    }
}

#[tokio::test]
async fn repos_by_owner_lists_in_path_order() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("multi", "multi@example.com"))
            .await
            .unwrap();
        for name in ["zeta", "alpha", "mu"] {
            fx.store
                .create_repo(NewRepo {
                    owner_id: owner.id.clone(),
                    group_id: None,
                    name: name.to_string(),
                    path: name.to_string(),
                    description: None,
                    visibility: model::Visibility::Private,
                    default_branch: "main".to_string(),
                })
                .await
                .unwrap();
        }
        let paths: Vec<String> = fx
            .store
            .repos_by_owner(&owner.id)
            .await
            .unwrap()
            .into_iter()
            .map(|r| r.path)
            .collect();
        assert_eq!(paths, ["alpha", "mu", "zeta"]);
    }
}

#[tokio::test]
async fn create_and_fetch_repo() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("owner", "owner@example.com"))
            .await
            .unwrap();

        let repo = fx
            .store
            .create_repo(NewRepo {
                owner_id: owner.id.clone(),
                group_id: None,
                name: "project".to_string(),
                path: "project".to_string(),
                description: Some("a repo".to_string()),
                visibility: model::Visibility::Public,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap();
        assert_eq!(repo.size_bytes, 0);
        assert!(repo.pushed_at.is_none());
        assert_eq!(repo.visibility, model::Visibility::Public);

        let fetched = fx
            .store
            .repo_by_owner_path(&owner.id, "project")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(fetched, repo);

        // Same owner, same path collides.
        let err = fx
            .store
            .create_repo(NewRepo {
                owner_id: owner.id.clone(),
                group_id: None,
                name: "project".to_string(),
                path: "project".to_string(),
                description: None,
                visibility: model::Visibility::Private,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap_err();
        assert!(
            matches!(
                err,
                StoreError::Conflict {
                    entity: "repo",
                    field: "path"
                }
            ),
            "got {err:?}"
        );
    }
}

#[tokio::test]
async fn foreign_keys_are_enforced() {
    // A repo whose owner does not exist must be rejected — proof the SQLite
    // `foreign_keys` pragma took effect on the pooled connection.
    for fx in sqlite_fixtures().await {
        let err = fx
            .store
            .create_repo(NewRepo {
                owner_id: "no-such-owner".to_string(),
                group_id: None,
                name: "orphan".to_string(),
                path: "orphan".to_string(),
                description: None,
                visibility: model::Visibility::Private,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap_err();
        assert!(
            matches!(err, StoreError::Query(_)),
            "expected a database (FK) error, got {err:?}"
        );
    }
}

#[test]
fn unsupported_url_is_rejected() {
    let err = Backend::detect("mysql://localhost/db").unwrap_err();
    assert!(
        matches!(err, StoreError::UnsupportedUrl { .. }),
        "got {err:?}"
    );
    assert_eq!(Backend::detect("sqlite::memory:").unwrap(), Backend::Sqlite);
    assert_eq!(
        Backend::detect("postgres://localhost/db").unwrap(),
        Backend::Postgres
    );
    assert_eq!(
        Backend::detect("postgresql://localhost/db").unwrap(),
        Backend::Postgres
    );
}

/// The two dialects' initial migrations must define the same tables with the
/// same columns in the same order — a hermetic check that needs no database.
/// The only sanctioned differences are the boolean column type and dialect
/// punctuation.
#[test]
fn schema_files_are_equivalent() {
    let sqlite = include_str!("../../../migrations/sqlite/0001_init.sql");
    let postgres = include_str!("../../../migrations/postgres/0001_init.sql");

    assert_eq!(
        columns_by_table(sqlite),
        columns_by_table(postgres),
        "SQLite and Postgres schemas must define identical tables and columns"
    );

    // The boolean columns must use the dialect-appropriate storage type.
    for column in ["must_change_password", "is_admin", "is_private"] {
        assert!(
            column_line(sqlite, column).contains("INTEGER"),
            "SQLite {column} should be INTEGER"
        );
        assert!(
            column_line(postgres, column).contains("BOOLEAN"),
            "Postgres {column} should be BOOLEAN"
        );
    }
}

/// Extract, per table, the ordered list of column names from a migration file.
/// Relies on the house style: each column on its own line beginning with the
/// column name, table-level constraints beginning with an uppercase keyword.
fn columns_by_table(sql: &str) -> std::collections::BTreeMap<String, Vec<String>> {
    const CONSTRAINT_KEYWORDS: &[&str] = &["UNIQUE", "PRIMARY", "FOREIGN", "CHECK", "CONSTRAINT"];
    let mut tables = std::collections::BTreeMap::new();
    let mut current: Option<(String, Vec<String>)> = None;

    for raw in sql.lines() {
        let line = raw.trim();
        if let Some(rest) = line.strip_prefix("CREATE TABLE ") {
            let name = rest.trim_end_matches(" (").trim_end_matches('(').trim();
            current = Some((name.to_string(), Vec::new()));
            continue;
        }
        let Some((name, cols)) = current.as_mut() else {
            continue;
        };
        if line.starts_with(')') {
            let (name, cols) = current.take().expect("current table");
            tables.insert(name, cols);
            continue;
        }
        // A column line starts with an identifier; skip blanks, comments, and
        // table-level constraints.
        let first = line.split([' ', '\t']).next().unwrap_or("");
        if first.is_empty()
            || line.starts_with("--")
            || CONSTRAINT_KEYWORDS.contains(&first.to_ascii_uppercase().as_str())
        {
            let _ = name;
            continue;
        }
        cols.push(first.to_string());
    }
    tables
}

/// Return the first line defining `column` (for type assertions).
fn column_line<'a>(sql: &'a str, column: &str) -> &'a str {
    sql.lines()
        .map(str::trim)
        .find(|line| line.split([' ', '\t']).next() == Some(column))
        .unwrap_or("")
}

#[tokio::test]
async fn ensure_group_path_creates_intermediates_and_reuses() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("g", "g@example.com"))
            .await
            .unwrap();

        let leaf = fx
            .store
            .ensure_group_path(&owner.id, "backend/service/v1")
            .await
            .unwrap();
        assert_eq!(leaf.name, "v1");
        assert_eq!(leaf.path, "backend/service/v1");

        // Every intermediate was created, parents linked.
        let all = fx.store.groups_by_owner(&owner.id).await.unwrap();
        let paths: Vec<&str> = all.iter().map(|g| g.path.as_str()).collect();
        assert_eq!(paths, ["backend", "backend/service", "backend/service/v1"]);
        let api = all.iter().find(|g| g.path == "backend/service").unwrap();
        let backend = all.iter().find(|g| g.path == "backend").unwrap();
        assert_eq!(api.parent_id.as_deref(), Some(backend.id.as_str()));
        assert!(backend.parent_id.is_none());

        // Re-ensuring reuses existing groups (no duplicates, same leaf id).
        let again = fx
            .store
            .ensure_group_path(&owner.id, "backend/service/v1")
            .await
            .unwrap();
        assert_eq!(again.id, leaf.id);
        assert_eq!(fx.store.groups_by_owner(&owner.id).await.unwrap().len(), 3);

        assert!(fx.store.delete_group(&backend.id).await.unwrap());
        // Deleting the top group cascades to its descendants.
        assert!(
            fx.store
                .groups_by_owner(&owner.id)
                .await
                .unwrap()
                .is_empty()
        );
    }
}

#[tokio::test]
async fn ensure_group_path_rejects_excessive_nesting() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("deep", "deep@example.com"))
            .await
            .unwrap();
        // Exactly MAX_GROUP_DEPTH levels is allowed.
        let at_limit = (1..=model::MAX_GROUP_DEPTH)
            .map(|n| format!("g{n}"))
            .collect::<Vec<_>>()
            .join("/");
        assert!(
            fx.store
                .ensure_group_path(&owner.id, &at_limit)
                .await
                .is_ok()
        );
        // One level deeper is rejected.
        let too_deep = format!("{at_limit}/extra");
        assert!(matches!(
            fx.store.ensure_group_path(&owner.id, &too_deep).await,
            Err(StoreError::GroupTooDeep { max }) if max == model::MAX_GROUP_DEPTH
        ));
    }
}

#[tokio::test]
async fn fork_parent_and_count_round_trip() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("fk", "fk@example.com"))
            .await
            .unwrap();
        let mk = |name: &str| NewRepo {
            owner_id: owner.id.clone(),
            group_id: None,
            name: name.to_string(),
            path: name.to_string(),
            description: None,
            visibility: model::Visibility::Public,
            default_branch: "main".to_string(),
        };
        let parent = fx.store.create_repo(mk("upstream")).await.unwrap();
        let fork = fx.store.create_repo(mk("myfork")).await.unwrap();
        assert!(fork.fork_parent_id.is_none());
        assert_eq!(fx.store.count_forks(&parent.id).await.unwrap(), 0);

        fx.store
            .set_fork_parent(&fork.id, &parent.id)
            .await
            .unwrap();
        let reloaded = fx.store.repo_by_id(&fork.id).await.unwrap().unwrap();
        assert_eq!(reloaded.fork_parent_id.as_deref(), Some(parent.id.as_str()));
        assert_eq!(fx.store.count_forks(&parent.id).await.unwrap(), 1);
    }
}

#[tokio::test]
async fn releases_and_assets_round_trip() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("rel", "rel@example.com"))
            .await
            .unwrap();
        let repo = fx
            .store
            .create_repo(NewRepo {
                owner_id: owner.id.clone(),
                group_id: None,
                name: "r".to_string(),
                path: "r".to_string(),
                description: None,
                visibility: model::Visibility::Public,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap();
        let rel = fx
            .store
            .create_release(crate::NewRelease {
                repo_id: repo.id.clone(),
                tag: "v1.0".to_string(),
                name: "First".to_string(),
                body: Some("notes".to_string()),
                is_prerelease: false,
                is_draft: false,
                author_id: owner.id.clone(),
            })
            .await
            .unwrap();
        // Duplicate tag conflicts.
        assert!(matches!(
            fx.store
                .create_release(crate::NewRelease {
                    repo_id: repo.id.clone(),
                    tag: "v1.0".to_string(),
                    name: "dup".to_string(),
                    body: None,
                    is_prerelease: false,
                    is_draft: false,
                    author_id: owner.id.clone(),
                })
                .await,
            Err(StoreError::Conflict { .. })
        ));
        assert_eq!(fx.store.list_releases(&repo.id).await.unwrap().len(), 1);
        assert_eq!(
            fx.store
                .release_by_tag(&repo.id, "v1.0")
                .await
                .unwrap()
                .unwrap()
                .id,
            rel.id
        );

        let asset = fx
            .store
            .add_release_asset(&rel.id, "bin.tar.gz", 4096, "application/gzip")
            .await
            .unwrap();
        assert_eq!(
            fx.store.list_release_assets(&rel.id).await.unwrap().len(),
            1
        );
        // Deleting the release cascades to its assets.
        assert!(fx.store.delete_release(&rel.id).await.unwrap());
        assert!(
            fx.store
                .release_asset_by_id(&asset.id)
                .await
                .unwrap()
                .is_none()
        );
    }
}

#[tokio::test]
async fn mirrors_create_list_due_and_delete() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("mir", "mir@example.com"))
            .await
            .unwrap();
        let repo = fx
            .store
            .create_repo(NewRepo {
                owner_id: owner.id,
                group_id: None,
                name: "r".to_string(),
                path: "r".to_string(),
                description: None,
                visibility: model::Visibility::Private,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap();
        let m = fx
            .store
            .create_mirror(crate::NewMirror {
                repo_id: repo.id.clone(),
                direction: crate::MirrorDirection::Push,
                remote_url: "https://example.com/r.git".to_string(),
                username: Some("u".to_string()),
                secret: Some("t".to_string()),
                branch_filter: None,
                interval_secs: 600,
                sync_on_push: true,
            })
            .await
            .unwrap();
        assert_eq!(fx.store.mirrors_for_repo(&repo.id).await.unwrap().len(), 1);
        // Never synced -> due now.
        let due = fx.store.due_mirrors(super::now_ms()).await.unwrap();
        assert_eq!(due.len(), 1);
        // sync-on-push mirrors are returned for the repo.
        assert_eq!(
            fx.store.push_mirrors_on_push(&repo.id).await.unwrap().len(),
            1
        );
        // After a successful sync, it is no longer due until the interval elapses.
        fx.store.mirror_synced(&m.id, None).await.unwrap();
        assert!(
            fx.store
                .due_mirrors(super::now_ms())
                .await
                .unwrap()
                .is_empty()
        );
        let stored = fx.store.mirror_by_id(&m.id).await.unwrap().unwrap();
        assert!(stored.last_sync_at.is_some() && stored.last_error.is_none());

        assert!(fx.store.delete_mirror(&m.id).await.unwrap());
        assert!(
            fx.store
                .mirrors_for_repo(&repo.id)
                .await
                .unwrap()
                .is_empty()
        );
    }
}

#[tokio::test]
async fn lfs_objects_track_per_repo() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("lfs", "lfs@example.com"))
            .await
            .unwrap();
        let repo = fx
            .store
            .create_repo(NewRepo {
                owner_id: owner.id,
                group_id: None,
                name: "r".to_string(),
                path: "r".to_string(),
                description: None,
                visibility: model::Visibility::Private,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap();
        let oid = "a".repeat(64);
        assert!(!fx.store.lfs_object_exists(&repo.id, &oid).await.unwrap());
        fx.store.lfs_object_add(&repo.id, &oid, 1234).await.unwrap();
        // A repeat add is a no-op (idempotent), not a conflict.
        fx.store.lfs_object_add(&repo.id, &oid, 1234).await.unwrap();
        assert!(fx.store.lfs_object_exists(&repo.id, &oid).await.unwrap());
        assert_eq!(
            fx.store.lfs_object_size(&repo.id, &oid).await.unwrap(),
            Some(1234)
        );
        assert_eq!(fx.store.lfs_repo_bytes(&repo.id).await.unwrap(), 1234);
    }
}

#[tokio::test]
async fn keys_get_stable_per_user_ordinals() {
    for fx in sqlite_fixtures().await {
        let user = fx
            .store
            .create_user(sample_user("k", "k@example.com"))
            .await
            .unwrap();

        let new_key = |fp: &str, kind| NewKey {
            user_id: user.id.clone(),
            kind,
            name: None,
            fingerprint: fp.to_string(),
            public_key: format!("material-{fp}"),
        };

        let k1 = fx
            .store
            .create_key(new_key("SHA256:a", KeyKind::Ssh))
            .await
            .unwrap();
        let k2 = fx
            .store
            .create_key(new_key("fp-b", KeyKind::Gpg))
            .await
            .unwrap();
        // Ordinals are unique per user across kinds, so `key del <index>` needs no
        // type.
        assert_eq!(k1.ordinal, 1);
        assert_eq!(k2.ordinal, 2);

        // Duplicate fingerprint is rejected.
        let err = fx
            .store
            .create_key(new_key("SHA256:a", KeyKind::Ssh))
            .await
            .unwrap_err();
        assert!(
            matches!(err, StoreError::Conflict { entity: "key", .. }),
            "got {err:?}"
        );

        // Delete the first; the second keeps its ordinal (no renumbering).
        assert!(fx.store.delete_key(&k1.id).await.unwrap());
        let remaining = fx.store.keys_by_user(&user.id).await.unwrap();
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].ordinal, 2);

        // A new key takes max(ordinal)+1, not a filled gap.
        let k3 = fx
            .store
            .create_key(new_key("SHA256:c", KeyKind::Ssh))
            .await
            .unwrap();
        assert_eq!(k3.ordinal, 3);

        let by_ord = fx.store.key_by_ordinal(&user.id, 2).await.unwrap().unwrap();
        assert_eq!(by_ord.id, k2.id);
        let by_fp = fx
            .store
            .key_by_fingerprint(KeyKind::Gpg, "fp-b")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(by_fp.id, k2.id);
    }
}

#[tokio::test]
async fn tokens_create_list_and_revoke() {
    for fx in sqlite_fixtures().await {
        let user = fx
            .store
            .create_user(sample_user("t", "t@example.com"))
            .await
            .unwrap();

        let first = fx
            .store
            .create_token(NewToken {
                user_id: user.id.clone(),
                name: "ci".to_string(),
                scopes: "repo:read".to_string(),
                expires_at: None,
            })
            .await
            .unwrap();
        fx.store
            .create_token(NewToken {
                user_id: user.id.clone(),
                name: "deploy".to_string(),
                scopes: "repo:read,repo:write".to_string(),
                expires_at: Some(now_plus_a_day()),
            })
            .await
            .unwrap();

        let listed = fx.store.tokens_by_user(&user.id).await.unwrap();
        assert_eq!(listed.len(), 2);
        assert_eq!(listed[0].name, "ci", "ordered by creation");

        // Revocation is what makes a token invalid to the API.
        assert!(fx.store.revoke_token(&first.id).await.unwrap());
        let revoked = fx.store.token_by_id(&first.id).await.unwrap().unwrap();
        assert!(revoked.revoked_at.is_some());
        // Revoking again is a no-op (already revoked).
        assert!(!fx.store.revoke_token(&first.id).await.unwrap());

        assert!(fx.store.delete_token(&first.id).await.unwrap());
        assert!(fx.store.token_by_id(&first.id).await.unwrap().is_none());
    }
}

fn now_plus_a_day() -> i64 {
    super::now_ms() + 86_400_000
}

#[tokio::test]
async fn rename_and_delete_repo() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("r", "r@example.com"))
            .await
            .unwrap();
        let repo = fx
            .store
            .create_repo(NewRepo {
                owner_id: owner.id.clone(),
                group_id: None,
                name: "old".to_string(),
                path: "old".to_string(),
                description: None,
                visibility: model::Visibility::Private,
                default_branch: "main".to_string(),
            })
            .await
            .unwrap();

        assert!(fx.store.rename_repo(&repo.id, "new", "new").await.unwrap());
        assert!(
            fx.store
                .repo_by_owner_path(&owner.id, "old")
                .await
                .unwrap()
                .is_none()
        );
        let renamed = fx.store.repo_by_id(&repo.id).await.unwrap().unwrap();
        assert_eq!(renamed.name, "new");
        assert_eq!(renamed.path, "new");

        assert!(fx.store.delete_repo(&repo.id).await.unwrap());
        assert!(fx.store.repo_by_id(&repo.id).await.unwrap().is_none());
        assert!(!fx.store.delete_repo(&repo.id).await.unwrap());
    }
}

#[tokio::test]
async fn sessions_resolve_the_user_and_respect_expiry() {
    use super::NewSession;

    for fx in sqlite_fixtures().await {
        let user = fx
            .store
            .create_user(sample_user("s", "s@example.com"))
            .await
            .unwrap();

        fx.store
            .create_session(NewSession {
                id: "sess-live".to_string(),
                user_id: user.id.clone(),
                user_agent: Some("curl".to_string()),
                ip: None,
                expires_at: now_plus_a_day(),
            })
            .await
            .unwrap();
        let resolved = fx.store.user_for_session("sess-live").await.unwrap();
        assert_eq!(resolved.map(|u| u.id), Some(user.id.clone()));

        // An expired session resolves to nobody and is reaped.
        fx.store
            .create_session(NewSession {
                id: "sess-dead".to_string(),
                user_id: user.id.clone(),
                user_agent: None,
                ip: None,
                expires_at: 1,
            })
            .await
            .unwrap();
        assert!(
            fx.store
                .user_for_session("sess-dead")
                .await
                .unwrap()
                .is_none()
        );
        assert_eq!(fx.store.delete_expired_sessions().await.unwrap(), 1);

        // Logout deletes the live session.
        assert!(fx.store.delete_session("sess-live").await.unwrap());
        assert!(
            fx.store
                .user_for_session("sess-live")
                .await
                .unwrap()
                .is_none()
        );
    }
}

#[tokio::test]
async fn invites_round_trip_and_are_single_use() {
    use super::NewInvite;

    for fx in sqlite_fixtures().await {
        let user = fx
            .store
            .create_user(sample_user("i", "i@example.com"))
            .await
            .unwrap();

        let invite = fx
            .store
            .create_invite(NewInvite {
                user_id: user.id.clone(),
                token_hash: "hash-abc".to_string(),
                purpose: "activate".to_string(),
                expires_at: now_plus_a_day(),
            })
            .await
            .unwrap();
        assert!(invite.used_at.is_none());

        let found = fx
            .store
            .invite_by_token_hash("hash-abc")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(found.user_id, user.id);
        assert_eq!(found.purpose, "activate");

        // Redemption is single-use: the first mark succeeds, the second does not.
        assert!(fx.store.mark_invite_used(&invite.id).await.unwrap());
        assert!(!fx.store.mark_invite_used(&invite.id).await.unwrap());
        let redeemed = fx
            .store
            .invite_by_token_hash("hash-abc")
            .await
            .unwrap()
            .unwrap();
        assert!(redeemed.used_at.is_some());

        assert!(
            fx.store
                .invite_by_token_hash("nope")
                .await
                .unwrap()
                .is_none()
        );
    }
}

/// Postgres round-trips, mirroring the SQLite ones. Ignored unless the
/// `postgres-tests` feature is on and `FABRICA_TEST_POSTGRES_URL` points at a
/// reachable database; the test cleans up the rows it creates so it can rerun.
#[tokio::test]
#[cfg_attr(
    not(feature = "postgres-tests"),
    ignore = "enable the postgres-tests feature and set FABRICA_TEST_POSTGRES_URL"
)]
async fn postgres_round_trip() {
    let Ok(url) = std::env::var("FABRICA_TEST_POSTGRES_URL") else {
        eprintln!("skipping postgres_round_trip: FABRICA_TEST_POSTGRES_URL not set");
        return;
    };

    let store = Store::connect(&url, 5).await.unwrap();
    assert_eq!(store.backend(), Backend::Postgres);
    store.migrate().await.unwrap();

    // Unique names so repeated runs against a shared database don't collide.
    let suffix = super::new_id();
    let username = format!("pg{suffix}");
    let email = format!("{suffix}@example.com");

    let admin = store
        .create_user(NewUser {
            is_admin: true,
            must_change_password: true,
            ..sample_user(&username, &email)
        })
        .await
        .unwrap();

    let fetched = store.user_by_id(&admin.id).await.unwrap().unwrap();
    assert_eq!(fetched, admin);
    assert!(fetched.is_admin, "native BOOLEAN must decode as true");
    assert!(fetched.must_change_password);

    let repo = store
        .create_repo(NewRepo {
            owner_id: admin.id.clone(),
            group_id: None,
            name: "pgproj".to_string(),
            path: "pgproj".to_string(),
            description: None,
            visibility: model::Visibility::Public,
            default_branch: "main".to_string(),
        })
        .await
        .unwrap();
    let fetched_repo = store
        .repo_by_owner_path(&admin.id, "pgproj")
        .await
        .unwrap()
        .unwrap();
    assert_eq!(fetched_repo, repo);
    assert_eq!(fetched_repo.visibility, model::Visibility::Public);

    // Clean up (repo cascades with the user).
    sqlx::query("DELETE FROM users WHERE id = $1")
        .bind(&admin.id)
        .execute(store.pool())
        .await
        .unwrap();
}

/// Helper: create a repo at `path`, auto-creating its group chain, and return it.
async fn repo_in_group(store: &Store, owner_id: &str, path: &str) -> model::Repo {
    let (group, leaf) = match path.rsplit_once('/') {
        Some((g, l)) => (Some(g), l),
        None => (None, path),
    };
    let group_id = match group {
        Some(g) => Some(store.ensure_group_path(owner_id, g).await.unwrap().id),
        None => None,
    };
    store
        .create_repo(NewRepo {
            owner_id: owner_id.to_string(),
            group_id,
            name: leaf.to_string(),
            path: path.to_string(),
            description: None,
            visibility: model::Visibility::Private,
            default_branch: "main".to_string(),
        })
        .await
        .unwrap()
}

#[tokio::test]
async fn deleting_last_repo_gcs_empty_auto_groups() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("gc", "gc@example.com"))
            .await
            .unwrap();
        // what/the/fuck where `fuck` is a repo; siblings keep their branch alive.
        let repo = repo_in_group(&fx.store, &owner.id, "what/the/fuck").await;
        let sibling = repo_in_group(&fx.store, &owner.id, "what/keep").await;

        assert!(
            fx.store
                .group_by_owner_path(&owner.id, "what/the")
                .await
                .unwrap()
                .is_some()
        );

        // Deleting `fuck` empties `the` (deleted), but `what` still holds `keep`.
        assert!(fx.store.delete_repo(&repo.id).await.unwrap());
        assert!(
            fx.store
                .group_by_owner_path(&owner.id, "what/the")
                .await
                .unwrap()
                .is_none(),
            "empty auto group `what/the` should be collected"
        );
        assert!(
            fx.store
                .group_by_owner_path(&owner.id, "what")
                .await
                .unwrap()
                .is_some(),
            "`what` still holds the `keep` repo"
        );

        // Removing the last repo collects the whole chain up to the root.
        assert!(fx.store.delete_repo(&sibling.id).await.unwrap());
        assert!(
            fx.store
                .group_by_owner_path(&owner.id, "what")
                .await
                .unwrap()
                .is_none(),
            "the now-empty root auto group is collected too"
        );
    }
}

#[tokio::test]
async fn manual_groups_survive_gc() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("man", "man@example.com"))
            .await
            .unwrap();
        // Manually create `team`, then a repo under an auto subgroup.
        let team = fx.store.create_group(&owner.id, "team").await.unwrap();
        assert!(team.manual);
        let repo = repo_in_group(&fx.store, &owner.id, "team/auto/proj").await;

        assert!(fx.store.delete_repo(&repo.id).await.unwrap());
        assert!(
            fx.store
                .group_by_owner_path(&owner.id, "team/auto")
                .await
                .unwrap()
                .is_none(),
            "auto subgroup collected"
        );
        assert!(
            fx.store
                .group_by_owner_path(&owner.id, "team")
                .await
                .unwrap()
                .is_some(),
            "manual group is never collected"
        );
    }
}

#[tokio::test]
async fn group_collaborator_grants_effective_permission() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("own", "own@example.com"))
            .await
            .unwrap();
        let collab = fx
            .store
            .create_user(sample_user("col", "col@example.com"))
            .await
            .unwrap();
        let repo = repo_in_group(&fx.store, &owner.id, "acme/deep/svc").await;

        // No access before any grant.
        assert!(
            fx.store
                .effective_permission(&repo.id, &collab.id)
                .await
                .unwrap()
                .is_none()
        );

        // Grant write on the top group; it cascades down to the nested repo.
        let acme = fx
            .store
            .group_by_owner_path(&owner.id, "acme")
            .await
            .unwrap()
            .unwrap();
        fx.store
            .add_group_collaborator(&acme.id, &collab.id, "write")
            .await
            .unwrap();
        assert_eq!(
            fx.store
                .effective_permission(&repo.id, &collab.id)
                .await
                .unwrap()
                .as_deref(),
            Some("write")
        );

        // A stronger repo-level grant wins over the weaker group grant.
        fx.store
            .set_collaborator(&repo.id, &collab.id, "admin")
            .await
            .unwrap();
        assert_eq!(
            fx.store
                .effective_permission(&repo.id, &collab.id)
                .await
                .unwrap()
                .as_deref(),
            Some("admin")
        );

        // The group-level effective permission also resolves for the group itself.
        let svc_group = fx
            .store
            .group_by_owner_path(&owner.id, "acme/deep")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            fx.store
                .effective_group_permission(&svc_group.id, &collab.id)
                .await
                .unwrap()
                .as_deref(),
            Some("write"),
            "ancestor group grant is inherited"
        );
    }
}

#[tokio::test]
async fn rename_group_moves_subtree() {
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("rn", "rn@example.com"))
            .await
            .unwrap();
        let repo = repo_in_group(&fx.store, &owner.id, "old/sub/proj").await;
        let top = fx
            .store
            .group_by_owner_path(&owner.id, "old")
            .await
            .unwrap()
            .unwrap();

        fx.store.rename_group(&top.id, "new").await.unwrap();

        // The repo and descendant group paths are rewritten under the new prefix.
        let moved = fx.store.repo_by_id(&repo.id).await.unwrap().unwrap();
        assert_eq!(moved.path, "new/sub/proj");
        assert!(
            fx.store
                .group_by_owner_path(&owner.id, "new/sub")
                .await
                .unwrap()
                .is_some()
        );
        assert!(
            fx.store
                .group_by_owner_path(&owner.id, "old/sub")
                .await
                .unwrap()
                .is_none()
        );
    }
}

#[tokio::test]
async fn group_visibility_ceiling_clamps_subtree() {
    use model::Visibility;
    for fx in sqlite_fixtures().await {
        let owner = fx
            .store
            .create_user(sample_user("vis", "vis@example.com"))
            .await
            .unwrap();
        let org = fx.store.create_group(&owner.id, "org").await.unwrap();
        let repo = repo_in_group(&fx.store, &owner.id, "org/team/app").await;
        // Start everything fully public.
        fx.store
            .set_visibility(&repo.id, Visibility::Public)
            .await
            .unwrap();
        let team = fx
            .store
            .group_by_owner_path(&owner.id, "org/team")
            .await
            .unwrap()
            .unwrap();
        fx.store
            .set_group_visibility(&team.id, Visibility::Public)
            .await
            .unwrap();
        assert_eq!(
            fx.store.group_visibility_ceiling(&team.id).await.unwrap(),
            Visibility::Public
        );

        // Lower the top group to internal and cascade the ceiling downward.
        fx.store
            .set_group_visibility(&org.id, Visibility::Internal)
            .await
            .unwrap();
        let org = fx.store.group_by_id(&org.id).await.unwrap().unwrap();
        fx.store.clamp_subtree_visibility(&org).await.unwrap();

        let team = fx
            .store
            .group_by_owner_path(&owner.id, "org/team")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            team.visibility,
            Visibility::Internal,
            "subgroup clamped to the new ceiling"
        );
        let repo = fx.store.repo_by_id(&repo.id).await.unwrap().unwrap();
        assert_eq!(
            repo.visibility,
            Visibility::Internal,
            "repo clamped to the new ceiling"
        );
        assert_eq!(
            fx.store.group_visibility_ceiling(&team.id).await.unwrap(),
            Visibility::Internal
        );
    }
}