knot-server 0.9.0

use axum::Json;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use std::sync::Arc;

use crate::handlers::models::*;
use crate::models::AppState;
use crate::models::{RegisterRepoRequest, RegisterRepoResponse, RepoEntry, RepoListResponse};

#[utoipa::path(
    get,
    path = "/api/repos",
    tag = "Repositories",
    responses(
        (status = 200, description = "List of all registered repositories", body = RepoListResponse),
    ),
    description = "List all registered Git repositories with their current status and metadata.",
)]
pub async fn list_repos_handler(State(state): State<Arc<AppState>>) -> Response {
    let mut registry = state.registry.lock().unwrap();
    let repos = registry.list().to_vec();
    let response = RepoListResponse {
        repositories: repos,
    };
    (StatusCode::OK, Json(response)).into_response()
}

#[utoipa::path(
    get,
    path = "/api/repos/{id}",
    tag = "Repositories",
    params(
        ("id" = String, Path, description = "Repository ID"),
    ),
    responses(
        (status = 200, description = "Repository details", body = RepoEntry),
        (status = 404, description = "Repository not found", body = ErrorResponse),
    ),
    description = "Get detailed information about a single registered repository.",
)]
pub async fn get_repo_handler(
    State(state): State<Arc<AppState>>,
    Path(id): Path<String>,
) -> Response {
    let mut registry = state.registry.lock().unwrap();
    match registry.get(&id) {
        Some(entry) => (StatusCode::OK, Json(entry.clone())).into_response(),
        None => error_response(
            StatusCode::NOT_FOUND,
            format!("Repository '{}' not found", id),
        ),
    }
}

#[utoipa::path(
    post,
    path = "/api/repos",
    tag = "Repositories",
    request_body = RegisterRepoRequest,
    responses(
        (status = 202, description = "Repository registered (or re-registered) and clone job enqueued", body = RegisterRepoResponse),
        (status = 400, description = "Empty URL, or a local path that does not exist", body = ErrorResponse),
        (status = 429, description = "Indexing queue is full", body = ErrorResponse),
    ),
    description = "Register a new Git repository, or re-register an existing one. The endpoint is idempotent: if a repository with the same derived id already exists, it is re-registered and a fresh Clone job is enqueued. The destructive cleanup (wiping the Neo4j/Qdrant data and the local directory) is performed by the indexing worker under its file lock — not by this endpoint — so a re-registration never races with in-flight indexing. Before wiping, the worker checks that the source is reachable (`git ls-remote` for a remote URL); if it is not, the job fails and a previously indexed repository keeps its index. A local path that does not exist is rejected with 400. The response message indicates whether the call was a fresh registration or a re-registration."
)]
#[tracing::instrument(
    name = "register_repo",
    skip_all,
    fields(repo_id = tracing::field::Empty)
)]
pub async fn register_repo_handler(
    State(state): State<Arc<AppState>>,
    Json(body): Json<RegisterRepoRequest>,
) -> Response {
    let id = match validate_register_request(&body) {
        Ok(id) => id,
        Err(e) => return e.into_response(),
    };

    let local_path = crate::models::repo_local_path(&state.workspace_dir, &id);

    let mut registry = state.registry.lock().unwrap();
    // A re-registration keeps `last_indexed`: the index survives until the
    // worker wipes it, and a job that fails before the wipe must not treat
    // the repository as never indexed (which would delete that index).
    let last_indexed = registry.get(&id).and_then(|e| e.last_indexed.clone());

    let entry = RepoEntry {
        id: id.clone(),
        url: body.url.clone(),
        auth_type: body.auth_type.clone(),
        local_path,
        branch: body.branch.clone(),
        webhook_secret: body.webhook_secret.clone(),
        last_indexed,
        status: crate::models::RepoStatus::Queued,
    };

    match registry.add_or_replace(entry) {
        Ok(was_replaced) => {
            // Enqueue a Clone job. The worker is the sole owner of destructive
            // cleanup: when it picks up this job it wipes the databases and the
            // local directory *under the file lock*, then clones from scratch
            // (see `worker::decide_job_plan`). The handler deliberately does NOT
            // touch `local_path` here — doing so in a background task raced with
            // the worker and left previously-healthy repos corrupted (Bug A).
            let job = crate::models::IndexJob::Clone {
                repo_id: id.clone(),
            };
            match state.job_tx.try_send(job) {
                Ok(()) => {}
                Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
                    let _ = registry.update_status(&id, crate::models::RepoStatus::Pending);
                    return error_response(
                        StatusCode::TOO_MANY_REQUESTS,
                        "Server is at maximum capacity: indexing queue is full",
                    );
                }
                Err(e) => {
                    tracing::error!("Failed to enqueue Clone job for {}: {e}", id);
                    let _ = registry.update_status(&id, crate::models::RepoStatus::Pending);
                }
            }

            if was_replaced {
                tracing::info!(
                    "Re-registered repository '{}' (url: {}, auth: {:?})",
                    id,
                    body.url,
                    body.auth_type
                );
            } else {
                tracing::info!(
                    "Registered repository '{}' (url: {}, auth: {:?})",
                    id,
                    body.url,
                    body.auth_type
                );
            }

            let response = RegisterRepoResponse {
                id,
                message: if was_replaced {
                    "Repository re-registered successfully".into()
                } else {
                    "Repository registered successfully".into()
                },
            };
            (StatusCode::ACCEPTED, Json(response)).into_response()
        }
        Err(e) => {
            let msg = e.to_string();
            error_response(StatusCode::INTERNAL_SERVER_ERROR, msg)
        }
    }
}

/// Validate a registration request and return the derived repository id, or
/// the 400 response to send.
fn validate_register_request(body: &RegisterRepoRequest) -> Result<String, HandlerError> {
    if body.url.trim().is_empty() {
        return Err(HandlerError::from(error_response(
            StatusCode::BAD_REQUEST,
            "Repository URL cannot be empty",
        )));
    }

    // Reject a missing local source up front: re-registering it would enqueue
    // a job that can only fail.
    if let Some(path) = crate::local_sync::local_source_path(&body.url)
        && !path.exists()
    {
        return Err(HandlerError::from(error_response(
            StatusCode::BAD_REQUEST,
            format!("Local repository path '{}' does not exist", path.display()),
        )));
    }

    let id = body.generate_id();
    tracing::Span::current().record("repo_id", id.as_str());
    if id.is_empty() {
        return Err(HandlerError::from(error_response(
            StatusCode::BAD_REQUEST,
            "Generated repository ID is empty (invalid URL format)",
        )));
    }
    Ok(id)
}

#[utoipa::path(
    delete,
    path = "/api/repos/{id}",
    tag = "Repositories",
    params(
        ("id" = String, Path, description = "Repository ID"),
    ),
    responses(
        (status = 200, description = "Repository deleted successfully", body = serde_json::Value),
        (status = 404, description = "Repository not found", body = ErrorResponse),
    ),
    description = "Delete a repository and clean up its databases and local files.",
)]
#[tracing::instrument(name = "delete_repo", skip_all, fields(repo_id = %id))]
pub async fn delete_repo_handler(
    State(state): State<Arc<AppState>>,
    Path(id): Path<String>,
) -> Response {
    // Scope the (non-Send) registry guard so it is fully released before the
    // `.await` below — otherwise the handler future is not `Send`/`Sync` and
    // fails axum's `Handler` bound.
    let removed = {
        let mut registry = state.registry.lock().unwrap();
        registry.remove(&id)
    };

    match removed {
        Ok(()) => {
            // Same cleanup the worker performs on a wipe, but for a full
            // delete: databases + local directory + progress snapshot/tracker.
            crate::cleanup::cleanup_repo_artifacts(
                &state,
                &id,
                crate::cleanup::CleanupScope {
                    databases: true,
                    local_dir: true,
                    progress: true,
                },
            )
            .await;
            remove_repo_lock_file(&state, &id);

            tracing::info!("Deleted repository '{}'", id);
            (
                StatusCode::OK,
                Json(serde_json::json!({"message": "Repository deleted"})),
            )
                .into_response()
        }
        Err(e) => {
            let msg = e.to_string();
            if msg.contains("not found") {
                error_response(StatusCode::NOT_FOUND, msg)
            } else {
                error_response(StatusCode::INTERNAL_SERVER_ERROR, msg)
            }
        }
    }
}

/// Remove the repository's lock file on delete. It lives outside the local
/// directory, so the cleanup above no longer takes it along; a leftover file
/// would keep the scheduler from picking up a re-registered `Pending` repo.
fn remove_repo_lock_file(state: &AppState, id: &str) {
    let lock_path = crate::locking::repo_lock_path(std::path::Path::new(&state.workspace_dir), id);
    if let Err(e) = std::fs::remove_file(&lock_path)
        && e.kind() != std::io::ErrorKind::NotFound
    {
        tracing::warn!("Failed to remove lock file {}: {e}", lock_path.display());
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::handlers::tests_common::create_test_state_with_rx;
    use crate::models::AuthType;
    use knot::pipeline::progress::ProgressTracker;
    use std::sync::Arc;
    use tempfile::TempDir;

    async fn create_test_state(workspace: &std::path::Path) -> Arc<AppState> {
        create_test_state_with_rx(workspace).await.0
    }

    #[tokio::test]
    async fn test_reregister_does_not_delete_local_dir_in_handler() {
        let dir = TempDir::new().unwrap();
        let workspace = dir.path().join("workspace");
        std::fs::create_dir_all(&workspace).unwrap();
        let (state, mut job_rx) = create_test_state_with_rx(&workspace).await;

        let url = "git@github.com:org/reg-test.git";
        let id = RegisterRepoRequest {
            url: url.into(),
            auth_type: AuthType::Ssh,
            branch: "main".into(),
            webhook_secret: None,
        }
        .generate_id();

        // Pre-existing, healthy local checkout: a .git dir and a file that a
        // stray background cleanup would have deleted.
        let local_path = crate::models::repo_local_path(&state.workspace_dir, &id);
        std::fs::create_dir_all(std::path::Path::new(&local_path).join(".git")).unwrap();
        std::fs::write(std::path::Path::new(&local_path).join("keep.txt"), b"keep").unwrap();

        // Register the same repo once so the second call is a re-registration.
        let entry = RepoEntry {
            id: id.clone(),
            url: url.into(),
            auth_type: AuthType::Ssh,
            local_path: local_path.clone(),
            branch: "main".into(),
            webhook_secret: None,
            last_indexed: Some("2026-07-06T00:00:00Z".into()),
            status: crate::models::RepoStatus::Indexed,
        };
        state
            .registry
            .lock()
            .unwrap()
            .add_or_replace(entry)
            .unwrap();

        let body = RegisterRepoRequest {
            url: url.into(),
            auth_type: AuthType::Ssh,
            branch: "main".into(),
            webhook_secret: None,
        };
        let response = register_repo_handler(State(state.clone()), Json(body)).await;
        assert_eq!(response.status(), StatusCode::ACCEPTED);

        // Give any (now-removed) background task a chance to run.
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        assert!(
            std::path::Path::new(&local_path).join("keep.txt").exists(),
            "handler must NOT delete the local dir on re-registration (worker does it)"
        );

        // Exactly one Clone job was enqueued.
        match job_rx.try_recv() {
            Ok(crate::models::IndexJob::Clone { repo_id }) => assert_eq!(repo_id, id),
            other => panic!("expected exactly one Clone job, got {other:?}"),
        }
        assert!(job_rx.try_recv().is_err(), "no extra jobs expected");
    }

    fn indexed_entry(id: &str, url: &str, local_path: &str) -> RepoEntry {
        RepoEntry {
            id: id.into(),
            url: url.into(),
            auth_type: AuthType::Ssh,
            local_path: local_path.into(),
            branch: "main".into(),
            webhook_secret: None,
            last_indexed: Some("2026-07-06T00:00:00Z".into()),
            status: crate::models::RepoStatus::Indexed,
        }
    }

    #[tokio::test]
    async fn test_register_missing_local_path_is_rejected_without_touching_entry() {
        let dir = TempDir::new().unwrap();
        let workspace = dir.path().join("workspace");
        std::fs::create_dir_all(&workspace).unwrap();
        let (state, mut job_rx) = create_test_state_with_rx(&workspace).await;

        let missing = dir.path().join("deleted").join("puppeteer");
        let missing_url = missing.to_string_lossy().into_owned();
        let local_path = crate::models::repo_local_path(&state.workspace_dir, "puppeteer");
        state
            .registry
            .lock()
            .unwrap()
            .add_or_replace(indexed_entry("puppeteer", &missing_url, &local_path))
            .unwrap();

        let body = RegisterRepoRequest {
            url: missing_url,
            auth_type: AuthType::Ssh,
            branch: "main".into(),
            webhook_secret: None,
        };
        let response = register_repo_handler(State(state.clone()), Json(body)).await;
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
        assert!(job_rx.try_recv().is_err(), "no job may be enqueued");

        let entry = state
            .registry
            .lock()
            .unwrap()
            .get("puppeteer")
            .unwrap()
            .clone();
        assert_eq!(entry.status, crate::models::RepoStatus::Indexed);
        assert!(entry.last_indexed.is_some());
    }

    #[tokio::test]
    async fn test_reregister_keeps_last_indexed() {
        let dir = TempDir::new().unwrap();
        let workspace = dir.path().join("workspace");
        std::fs::create_dir_all(&workspace).unwrap();
        let (state, _job_rx) = create_test_state_with_rx(&workspace).await;

        let url = "git@github.com:org/kept.git";
        let local_path = crate::models::repo_local_path(&state.workspace_dir, "kept");
        state
            .registry
            .lock()
            .unwrap()
            .add_or_replace(indexed_entry("kept", url, &local_path))
            .unwrap();

        let body = RegisterRepoRequest {
            url: url.into(),
            auth_type: AuthType::Ssh,
            branch: "main".into(),
            webhook_secret: None,
        };
        let response = register_repo_handler(State(state.clone()), Json(body)).await;
        assert_eq!(response.status(), StatusCode::ACCEPTED);

        let entry = state.registry.lock().unwrap().get("kept").unwrap().clone();
        assert_eq!(entry.status, crate::models::RepoStatus::Queued);
        assert_eq!(entry.last_indexed.as_deref(), Some("2026-07-06T00:00:00Z"));
    }

    #[tokio::test]
    async fn test_delete_removes_tracker_from_state() {
        let dir = TempDir::new().unwrap();
        let workspace = dir.path().join("workspace");
        std::fs::create_dir_all(&workspace).unwrap();

        let state = create_test_state(&workspace).await;

        let entry = RepoEntry {
            id: "delete-test".into(),
            url: "git@github.com:org/delete-test.git".into(),
            local_path: "/tmp/delete-test".into(),
            auth_type: AuthType::Ssh,
            branch: "main".into(),
            webhook_secret: None,
            last_indexed: None,
            status: crate::models::RepoStatus::Indexed,
        };
        state
            .registry
            .lock()
            .unwrap()
            .add_or_replace(entry)
            .unwrap();

        state
            .progress_trackers
            .lock()
            .unwrap()
            .insert("delete-test".into(), Arc::new(ProgressTracker::new()));

        let response = delete_repo_handler(State(state.clone()), Path("delete-test".into())).await;

        assert_eq!(response.status(), StatusCode::OK);

        let map = state.progress_trackers.lock().unwrap();
        assert!(
            !map.contains_key("delete-test"),
            "tracker should be removed after delete"
        );
    }

    #[tokio::test]
    async fn test_delete_removes_repo_lock_file() {
        let dir = TempDir::new().unwrap();
        let workspace = dir.path().join("workspace");
        std::fs::create_dir_all(&workspace).unwrap();

        let state = create_test_state(&workspace).await;

        let entry = RepoEntry {
            id: "lock-test".into(),
            url: "git@github.com:org/lock-test.git".into(),
            local_path: workspace.join("lock-test").to_string_lossy().into_owned(),
            auth_type: AuthType::Ssh,
            branch: "main".into(),
            webhook_secret: None,
            last_indexed: None,
            status: crate::models::RepoStatus::Indexed,
        };
        state
            .registry
            .lock()
            .unwrap()
            .add_or_replace(entry)
            .unwrap();

        let lock_path = crate::locking::repo_lock_path(&workspace, "lock-test");
        drop(crate::locking::acquire_file_lock(&lock_path).unwrap());
        assert!(lock_path.exists());

        let response = delete_repo_handler(State(state.clone()), Path("lock-test".into())).await;

        assert_eq!(response.status(), StatusCode::OK);
        assert!(
            !lock_path.exists(),
            "lock file should be removed after delete"
        );
    }
}