termi-blog/backend/src/controllers/admin_ops.rs

use axum::http::HeaderMap;
use loco_rs::prelude::*;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, Order, QueryFilter, QueryOrder,
    QuerySelect, Set,
};
use serde::{Deserialize, Serialize};

use crate::{
    controllers::admin::check_auth,
    models::_entities::{admin_audit_logs, notification_deliveries, post_revisions, subscriptions},
    services::{
        admin_audit, backups, post_revisions as revision_service,
        subscriptions as subscription_service, worker_jobs,
    },
};

#[derive(Clone, Debug, Default, Deserialize)]
pub struct AuditLogQuery {
    pub action: Option<String>,
    pub target_type: Option<String>,
    pub limit: Option<u64>,
}

#[derive(Clone, Debug, Default, Deserialize)]
pub struct RevisionQuery {
    pub slug: Option<String>,
    pub limit: Option<u64>,
}

#[derive(Clone, Debug, Default, Deserialize)]
pub struct DeliveriesQuery {
    pub limit: Option<u64>,
}

#[derive(Clone, Debug, Default, Deserialize)]
pub struct WorkerJobsQuery {
    pub status: Option<String>,
    pub job_kind: Option<String>,
    pub worker_name: Option<String>,
    pub search: Option<String>,
    pub limit: Option<u64>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct SubscriptionPayload {
    #[serde(alias = "channelType")]
    pub channel_type: String,
    pub target: String,
    #[serde(default, alias = "displayName")]
    pub display_name: Option<String>,
    #[serde(default)]
    pub status: Option<String>,
    #[serde(default)]
    pub filters: Option<serde_json::Value>,
    #[serde(default)]
    pub metadata: Option<serde_json::Value>,
    #[serde(default)]
    pub secret: Option<String>,
    #[serde(default)]
    pub notes: Option<String>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct SubscriptionUpdatePayload {
    #[serde(default, alias = "channelType")]
    pub channel_type: Option<String>,
    #[serde(default)]
    pub target: Option<String>,
    #[serde(default, alias = "displayName")]
    pub display_name: Option<String>,
    #[serde(default)]
    pub status: Option<String>,
    #[serde(default)]
    pub filters: Option<serde_json::Value>,
    #[serde(default)]
    pub metadata: Option<serde_json::Value>,
    #[serde(default)]
    pub secret: Option<String>,
    #[serde(default)]
    pub notes: Option<String>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct RestoreRevisionRequest {
    #[serde(default)]
    pub mode: Option<String>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct DigestDispatchRequest {
    pub period: Option<String>,
}

#[derive(Clone, Debug, Default, Deserialize)]
pub struct RetryDeliveriesRequest {
    pub limit: Option<u64>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct SiteBackupImportRequest {
    pub backup: backups::SiteBackupDocument,
    #[serde(default)]
    pub mode: Option<String>,
}

#[derive(Clone, Debug, Serialize)]
pub struct PostRevisionListItem {
    pub id: i32,
    pub post_slug: String,
    pub post_title: Option<String>,
    pub operation: String,
    pub revision_reason: Option<String>,
    pub actor_username: Option<String>,
    pub actor_email: Option<String>,
    pub actor_source: Option<String>,
    pub created_at: String,
    pub has_markdown: bool,
    pub metadata: Option<serde_json::Value>,
}

#[derive(Clone, Debug, Serialize)]
pub struct PostRevisionDetailResponse {
    #[serde(flatten)]
    pub item: PostRevisionListItem,
    pub markdown: Option<String>,
}

#[derive(Clone, Debug, Serialize)]
pub struct RestoreRevisionResponse {
    pub restored: bool,
    pub revision_id: i32,
    pub post_slug: String,
    pub mode: String,
}

#[derive(Clone, Debug, Serialize)]
pub struct SubscriptionListResponse {
    pub subscriptions: Vec<subscriptions::Model>,
}

#[derive(Clone, Debug, Serialize)]
pub struct DeliveryListResponse {
    pub deliveries: Vec<notification_deliveries::Model>,
}

#[derive(Clone, Debug, Serialize)]
pub struct WorkerTaskActionResponse {
    pub queued: bool,
    pub job: worker_jobs::WorkerJobRecord,
}

fn trim_to_option(value: Option<String>) -> Option<String> {
    value.and_then(|item| {
        let trimmed = item.trim().to_string();
        if trimmed.is_empty() {
            None
        } else {
            Some(trimmed)
        }
    })
}

fn format_revision(item: post_revisions::Model) -> PostRevisionListItem {
    PostRevisionListItem {
        id: item.id,
        post_slug: item.post_slug,
        post_title: item.post_title,
        operation: item.operation,
        revision_reason: item.revision_reason,
        actor_username: item.actor_username,
        actor_email: item.actor_email,
        actor_source: item.actor_source,
        created_at: item.created_at.format("%Y-%m-%d %H:%M:%S").to_string(),
        has_markdown: item
            .markdown
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .is_some(),
        metadata: item.metadata,
    }
}

#[debug_handler]
pub async fn list_audit_logs(
    headers: HeaderMap,
    Query(query): Query<AuditLogQuery>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    check_auth(&headers)?;

    let mut db_query =
        admin_audit_logs::Entity::find().order_by(admin_audit_logs::Column::CreatedAt, Order::Desc);

    if let Some(action) = query
        .action
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
    {
        db_query = db_query.filter(admin_audit_logs::Column::Action.eq(action));
    }

    if let Some(target_type) = query
        .target_type
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
    {
        db_query = db_query.filter(admin_audit_logs::Column::TargetType.eq(target_type));
    }

    format::json(
        db_query
            .limit(query.limit.unwrap_or(80))
            .all(&ctx.db)
            .await?,
    )
}

#[debug_handler]
pub async fn list_post_revisions(
    headers: HeaderMap,
    Query(query): Query<RevisionQuery>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    check_auth(&headers)?;
    let items =
        revision_service::list_revisions(&ctx, query.slug.as_deref(), query.limit.unwrap_or(120))
            .await?;
    format::json(items.into_iter().map(format_revision).collect::<Vec<_>>())
}

#[debug_handler]
pub async fn get_post_revision(
    headers: HeaderMap,
    Path(id): Path<i32>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    check_auth(&headers)?;
    let item = revision_service::get_revision(&ctx, id).await?;
    format::json(PostRevisionDetailResponse {
        item: format_revision(item.clone()),
        markdown: item.markdown,
    })
}

#[debug_handler]
pub async fn restore_post_revision(
    headers: HeaderMap,
    Path(id): Path<i32>,
    State(ctx): State<AppContext>,
    Json(payload): Json<RestoreRevisionRequest>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;
    let mode = payload.mode.unwrap_or_else(|| "full".to_string());
    let restored = revision_service::restore_revision(&ctx, Some(&actor), id, &mode).await?;
    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "post.revision.restore",
        "post_revision",
        Some(restored.id.to_string()),
        Some(restored.post_slug.clone()),
        Some(serde_json::json!({
            "post_slug": restored.post_slug,
            "source_revision_id": id,
            "mode": mode,
        })),
    )
    .await?;

    format::json(RestoreRevisionResponse {
        restored: true,
        revision_id: id,
        post_slug: restored.post_slug,
        mode,
    })
}

#[debug_handler]
pub async fn list_subscriptions(
    headers: HeaderMap,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    check_auth(&headers)?;
    format::json(SubscriptionListResponse {
        subscriptions: subscription_service::list_subscriptions(&ctx, None, None).await?,
    })
}

#[debug_handler]
pub async fn list_subscription_deliveries(
    headers: HeaderMap,
    Query(query): Query<DeliveriesQuery>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    check_auth(&headers)?;
    format::json(DeliveryListResponse {
        deliveries: subscription_service::list_recent_deliveries(&ctx, query.limit.unwrap_or(80))
            .await?,
    })
}

#[debug_handler]
pub async fn create_subscription(
    headers: HeaderMap,
    State(ctx): State<AppContext>,
    Json(payload): Json<SubscriptionPayload>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;

    let channel_type = subscription_service::normalize_channel_type(&payload.channel_type);
    let target = payload.target.trim().to_string();
    if target.is_empty() {
        return Err(Error::BadRequest("target 不能为空".to_string()));
    }

    let created = subscriptions::ActiveModel {
        channel_type: Set(channel_type.clone()),
        target: Set(target.clone()),
        display_name: Set(trim_to_option(payload.display_name)),
        status: Set(subscription_service::normalize_status(
            payload.status.as_deref().unwrap_or("active"),
        )),
        filters: Set(subscription_service::normalize_filters(payload.filters)),
        metadata: Set(payload.metadata),
        secret: Set(trim_to_option(payload.secret)),
        notes: Set(trim_to_option(payload.notes)),
        confirm_token: Set(None),
        manage_token: Set(Some(subscription_service::generate_subscription_token())),
        verified_at: Set(Some(chrono::Utc::now().to_rfc3339())),
        failure_count: Set(Some(0)),
        ..Default::default()
    }
    .insert(&ctx.db)
    .await?;

    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "subscription.create",
        "subscription",
        Some(created.id.to_string()),
        Some(format!("{}:{}", created.channel_type, created.target)),
        Some(serde_json::json!({ "channel_type": created.channel_type, "target": created.target })),
    )
    .await?;

    format::json(created)
}

#[debug_handler]
pub async fn update_subscription(
    headers: HeaderMap,
    Path(id): Path<i32>,
    State(ctx): State<AppContext>,
    Json(payload): Json<SubscriptionUpdatePayload>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;

    let item = subscriptions::Entity::find_by_id(id)
        .one(&ctx.db)
        .await?
        .ok_or(Error::NotFound)?;
    let mut active = item.clone().into_active_model();

    if let Some(channel_type) = payload.channel_type {
        active.channel_type = Set(subscription_service::normalize_channel_type(&channel_type));
    }
    if let Some(target) = payload.target {
        let normalized_target = target.trim().to_string();
        if normalized_target.is_empty() {
            return Err(Error::BadRequest("target 不能为空".to_string()));
        }
        active.target = Set(normalized_target);
    }
    if payload.display_name.is_some() {
        active.display_name = Set(trim_to_option(payload.display_name));
    }
    if let Some(status) = payload.status {
        active.status = Set(subscription_service::normalize_status(&status));
    }
    if payload.filters.is_some() {
        active.filters = Set(subscription_service::normalize_filters(payload.filters));
    }
    if payload.metadata.is_some() {
        active.metadata = Set(payload.metadata);
    }
    if payload.secret.is_some() {
        active.secret = Set(trim_to_option(payload.secret));
    }
    if payload.notes.is_some() {
        active.notes = Set(trim_to_option(payload.notes));
    }

    let updated = active.update(&ctx.db).await?;
    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "subscription.update",
        "subscription",
        Some(updated.id.to_string()),
        Some(format!("{}:{}", updated.channel_type, updated.target)),
        None,
    )
    .await?;

    format::json(updated)
}

#[debug_handler]
pub async fn delete_subscription(
    headers: HeaderMap,
    Path(id): Path<i32>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;
    let item = subscriptions::Entity::find_by_id(id)
        .one(&ctx.db)
        .await?
        .ok_or(Error::NotFound)?;
    let label = format!("{}:{}", item.channel_type, item.target);
    item.delete(&ctx.db).await?;

    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "subscription.delete",
        "subscription",
        Some(id.to_string()),
        Some(label),
        None,
    )
    .await?;

    format::empty()
}

#[debug_handler]
pub async fn test_subscription(
    headers: HeaderMap,
    Path(id): Path<i32>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;
    let item = subscriptions::Entity::find_by_id(id)
        .one(&ctx.db)
        .await?
        .ok_or(Error::NotFound)?;

    let delivery = subscription_service::send_test_notification(&ctx, &item).await?;
    let job = worker_jobs::find_latest_job_by_related_entity(
        &ctx,
        "notification_delivery",
        &delivery.id.to_string(),
        Some(worker_jobs::WORKER_NOTIFICATION_DELIVERY),
    )
    .await?;
    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "subscription.test",
        "subscription",
        Some(item.id.to_string()),
        Some(format!("{}:{}", item.channel_type, item.target)),
        None,
    )
    .await?;

    format::json(serde_json::json!({
        "queued": true,
        "id": item.id,
        "delivery_id": delivery.id,
        "job_id": job.as_ref().map(|value| value.id),
    }))
}

#[debug_handler]
pub async fn send_subscription_digest(
    headers: HeaderMap,
    State(ctx): State<AppContext>,
    Json(payload): Json<DigestDispatchRequest>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;
    let summary =
        subscription_service::send_digest(&ctx, payload.period.as_deref().unwrap_or("weekly"))
            .await?;

    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "subscription.digest.send",
        "subscription_digest",
        None,
        Some(summary.period.clone()),
        Some(serde_json::json!({
            "period": summary.period,
            "post_count": summary.post_count,
            "queued": summary.queued,
            "skipped": summary.skipped,
        })),
    )
    .await?;

    format::json(summary)
}

#[debug_handler]
pub async fn workers_overview(
    headers: HeaderMap,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    check_auth(&headers)?;
    format::json(worker_jobs::get_overview(&ctx).await?)
}

#[debug_handler]
pub async fn list_worker_jobs(
    headers: HeaderMap,
    Query(query): Query<WorkerJobsQuery>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    check_auth(&headers)?;
    format::json(
        worker_jobs::list_jobs(
            &ctx,
            worker_jobs::WorkerJobListQuery {
                status: query.status,
                job_kind: query.job_kind,
                worker_name: query.worker_name,
                search: query.search,
                limit: query.limit,
            },
        )
        .await?,
    )
}

#[debug_handler]
pub async fn get_worker_job(
    headers: HeaderMap,
    Path(id): Path<i32>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    check_auth(&headers)?;
    format::json(worker_jobs::get_job_record(&ctx, id).await?)
}

#[debug_handler]
pub async fn cancel_worker_job(
    headers: HeaderMap,
    Path(id): Path<i32>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;
    let updated = worker_jobs::request_cancel(&ctx, id).await?;

    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "worker.cancel",
        "worker_job",
        Some(id.to_string()),
        Some(updated.worker_name.clone()),
        Some(serde_json::json!({ "status": updated.status })),
    )
    .await?;

    format::json(updated)
}

#[debug_handler]
pub async fn retry_worker_job(
    headers: HeaderMap,
    Path(id): Path<i32>,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;
    let job = worker_jobs::retry_job(
        &ctx,
        id,
        Some(actor.username.clone()),
        Some(actor.source.clone()),
    )
    .await?;

    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "worker.retry",
        "worker_job",
        Some(job.id.to_string()),
        Some(job.worker_name.clone()),
        Some(serde_json::json!({ "source_job_id": id })),
    )
    .await?;

    format::json(WorkerTaskActionResponse { queued: true, job })
}

#[debug_handler]
pub async fn run_retry_deliveries_job(
    headers: HeaderMap,
    State(ctx): State<AppContext>,
    Json(payload): Json<RetryDeliveriesRequest>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;
    let job = worker_jobs::spawn_retry_deliveries_task(
        &ctx,
        payload.limit,
        Some(actor.username.clone()),
        Some(actor.source.clone()),
        None,
        Some("manual".to_string()),
    )
    .await?;

    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "worker.task.retry_deliveries",
        "worker_job",
        Some(job.id.to_string()),
        Some(job.worker_name.clone()),
        Some(serde_json::json!({ "limit": payload.limit })),
    )
    .await?;

    format::json(WorkerTaskActionResponse { queued: true, job })
}

#[debug_handler]
pub async fn run_digest_worker_job(
    headers: HeaderMap,
    State(ctx): State<AppContext>,
    Json(payload): Json<DigestDispatchRequest>,
) -> Result<Response> {
    let actor = check_auth(&headers)?;
    let period = payload.period.unwrap_or_else(|| "weekly".to_string());
    let job = worker_jobs::spawn_digest_task(
        &ctx,
        &period,
        Some(actor.username.clone()),
        Some(actor.source.clone()),
        None,
        Some("manual".to_string()),
    )
    .await?;

    admin_audit::log_event(
        &ctx,
        Some(&actor),
        "worker.task.digest",
        "worker_job",
        Some(job.id.to_string()),
        Some(job.worker_name.clone()),
        Some(serde_json::json!({ "period": period })),
    )
    .await?;

    format::json(WorkerTaskActionResponse { queued: true, job })
}

#[debug_handler]
pub async fn export_site_backup(
    headers: HeaderMap,
    State(ctx): State<AppContext>,
) -> Result<Response> {
    check_auth(&headers)?;
    format::json(backups::export_site_backup(&ctx).await?)
}

#[debug_handler]
pub async fn import_site_backup(
    headers: HeaderMap,
    State(ctx): State<AppContext>,
    Json(payload): Json<SiteBackupImportRequest>,
) -> Result<Response> {
    check_auth(&headers)?;
    format::json(backups::import_site_backup(&ctx, payload.backup, payload.mode.as_deref()).await?)
}

pub fn routes() -> Routes {
    Routes::new()
        .prefix("/api/admin")
        .add("/audit-logs", get(list_audit_logs))
        .add("/post-revisions", get(list_post_revisions))
        .add("/post-revisions/{id}", get(get_post_revision))
        .add("/post-revisions/{id}/restore", post(restore_post_revision))
        .add(
            "/subscriptions",
            get(list_subscriptions).post(create_subscription),
        )
        .add(
            "/subscriptions/deliveries",
            get(list_subscription_deliveries),
        )
        .add("/subscriptions/digest", post(send_subscription_digest))
        .add(
            "/subscriptions/{id}",
            patch(update_subscription).delete(delete_subscription),
        )
        .add("/subscriptions/{id}/test", post(test_subscription))
        .add("/workers/overview", get(workers_overview))
        .add("/workers/jobs", get(list_worker_jobs))
        .add("/workers/jobs/{id}", get(get_worker_job))
        .add("/workers/jobs/{id}/cancel", post(cancel_worker_job))
        .add("/workers/jobs/{id}/retry", post(retry_worker_job))
        .add(
            "/workers/tasks/retry-deliveries",
            post(run_retry_deliveries_job),
        )
        .add("/workers/tasks/digest", post(run_digest_worker_job))
        .add("/site-backup/export", get(export_site_backup))
        .add("/site-backup/import", post(import_site_backup))
}