//! SyncKit group management and group-scoped push/pull.
//!
//! A group is a shared, end-to-end-encrypted changelog. The admin mints a Group
//! Content Key (GCK) client-side and seals it to each member's identity public
//! key; the server stores membership and the opaque sealed grants and never sees
//! the GCK or any plaintext. Group push/pull are gated on membership; management
//! actions (add/remove member) are gated on being the group admin.
//!
//! Member public keys are stored on add, and GCK rotation is exposed at
//! `/groups/{id}/rotate`: the grant set the admin posts becomes the new
//! membership, so removing a member and re-keying the group are one transaction.
//!
//! The paid-write gate is implemented: `group_push` checks the group admin's
//! entitlement and answers 402 `no_subscription`. Still deferred, to p3: SSE
//! push notifications for groups, noted where they would hook in. Design: wiki
//! synckit-groups-design.

use axum::{
    Json,
    extract::{Path, Query, State},
    http::StatusCode,
    response::{IntoResponse, Response},
};
use chrono::{Duration, Utc};
use serde_json::json;
use sqlx::PgPool;

use crate::{
    constants,
    db::{self, DbSyncGroup, SyncGroupId, SyncGroupInvitationId, UserId},
    error::{AppError, Result},
    synckit_auth::SyncUser,
    validation,
};

use super::{
    AcceptInvitationRequest, AddMemberRequest, ConfirmInvitationRequest, CreateGroupRequest,
    CreateInvitationRequest, CreateInvitationResponse, GrantQuery, GroupGrantResponse,
    GroupMemberPubkey, GroupMemberResponse, GroupResponse, InvitationPreviewResponse,
    InvitationResponse, PullChangeEntry, PullRequest, PullResponse, PushRequest, PushResponse,
    RotateGroupKeyRequest,
};

/// Fetch a group scoped to the caller's app, or 404. Guards every group handler
/// against cross-app id guessing before any membership check.
async fn require_group(
    db: &PgPool,
    app_id: db::SyncAppId,
    group_id: SyncGroupId,
) -> Result<DbSyncGroup> {
    db::synckit::get_group(db, app_id, group_id)
        .await?
        .ok_or(AppError::NotFound)
}

/// Reject a caller who is not a member of the group (403). The gate for
/// group-scoped reads and writes.
async fn require_member(db: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<()> {
    if db::synckit::is_group_member(db, group_id, user_id).await? {
        Ok(())
    } else {
        Err(AppError::Forbidden)
    }
}

/// Reject a caller who is not the group admin (403). The gate for membership
/// management.
async fn require_admin(db: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<()> {
    if db::synckit::is_group_admin(db, group_id, user_id).await? {
        Ok(())
    } else {
        Err(AppError::Forbidden)
    }
}

/// Create a group. The caller becomes its admin and first member, carrying the
/// GCK they sealed to their own identity key.
#[utoipa::path(post, path = "/api/v1/sync/groups", tag = "SyncKit",
    request_body = CreateGroupRequest,
    responses((status = 200, description = "Created group", body = GroupResponse)),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::create_group")]
pub(super) async fn create_group(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Json(req): Json<CreateGroupRequest>,
) -> Result<impl IntoResponse> {
    validation::validate_sync_group_name(&req.name)?;
    if req.admin_sealed_gck.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES
        || req.admin_pubkey.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES
    {
        return Err(AppError::BadRequest(
            "Sealed key exceeds size limit".to_string(),
        ));
    }

    let group = db::synckit::create_group(
        &db,
        req.id,
        sync_user.app_id,
        sync_user.user_id,
        &req.name,
        &req.admin_sealed_gck,
        &req.admin_pubkey,
    )
    .await?;

    Ok(Json(GroupResponse::from(group)))
}

/// List the groups the caller belongs to within this app.
#[utoipa::path(get, path = "/api/v1/sync/groups", tag = "SyncKit",
    responses((status = 200, description = "Groups the user belongs to", body = Vec<GroupResponse>)),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::list_groups")]
pub(super) async fn list_groups(
    State(db): State<PgPool>,
    sync_user: SyncUser,
) -> Result<impl IntoResponse> {
    let groups =
        db::synckit::list_groups_for_user(&db, sync_user.app_id, sync_user.user_id).await?;
    let response: Vec<GroupResponse> = groups.into_iter().map(GroupResponse::from).collect();
    Ok(Json(response))
}

/// Add a member to a group (or replace their grant). Admin only.
///
/// The admin resolves the member out of band, seals the current GCK to that
/// member's public key, and posts `{member_email, sealed_gck}`. The server maps
/// the email to a verified account and stores the opaque grant at the group's
/// current GCK generation.
#[utoipa::path(post, path = "/api/v1/sync/groups/{id}/members", tag = "SyncKit",
    params(("id" = String, Path, description = "Group ID")),
    request_body = AddMemberRequest,
    responses((status = 204, description = "Member added"), (status = 403, description = "Not the group admin")),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::add_group_member")]
pub(super) async fn add_member(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path(group_id): Path<SyncGroupId>,
    Json(req): Json<AddMemberRequest>,
) -> Result<impl IntoResponse> {
    let group = require_group(&db, sync_user.app_id, group_id).await?;
    require_admin(&db, group_id, sync_user.user_id).await?;

    if req.sealed_gck.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES
        || req.member_pubkey.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES
    {
        return Err(AppError::BadRequest(
            "Sealed key exceeds size limit".to_string(),
        ));
    }
    let role = req.role.as_deref().unwrap_or("member");
    if role != "member" && role != "admin" {
        return Err(AppError::BadRequest(
            "role must be 'member' or 'admin'".to_string(),
        ));
    }

    let email = db::Email::new(&req.member_email)
        .map_err(|_| AppError::BadRequest("Invalid email address".to_string()))?;
    let member_id = db::users::get_verified_user_id_by_email(&db, &email)
        .await?
        .ok_or_else(|| AppError::BadRequest("No verified account for that email".to_string()))?;

    // The grant the admin sends is sealed under the group's current GCK, so it is
    // stored at that generation, along with the member's public key (for re-seal
    // on a later rotation).
    db::synckit::add_or_update_member(
        &db,
        group_id,
        member_id,
        role,
        &req.sealed_gck,
        group.gck_version,
        &req.member_pubkey,
    )
    .await?;

    Ok(StatusCode::NO_CONTENT)
}

/// List every member's identity public key. Admin only: the admin re-seals a
/// rotated GCK to each of these on member removal.
#[utoipa::path(get, path = "/api/v1/sync/groups/{id}/pubkeys", tag = "SyncKit",
    params(("id" = String, Path, description = "Group ID")),
    responses((status = 200, description = "Member public keys", body = Vec<GroupMemberPubkey>)),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::list_group_pubkeys")]
pub(super) async fn list_pubkeys(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path(group_id): Path<SyncGroupId>,
) -> Result<impl IntoResponse> {
    require_group(&db, sync_user.app_id, group_id).await?;
    require_admin(&db, group_id, sync_user.user_id).await?;

    let pubkeys = db::synckit::list_member_pubkeys(&db, group_id).await?;
    let response: Vec<GroupMemberPubkey> = pubkeys
        .into_iter()
        .map(|(user_id, pubkey)| GroupMemberPubkey { user_id, pubkey })
        .collect();
    Ok(Json(response))
}

/// Rotate the group's Group Content Key. Admin only.
///
/// The admin mints a fresh GCK client-side, seals it to each remaining member's
/// stored public key (from `/pubkeys`), and posts the batch. The server bumps the
/// generation, drops anyone absent from the batch, and stores the new grants in
/// one transaction.
///
/// The grant set IS the new membership, which is what makes removal and re-key
/// atomic: there is no window in which a removed member's key is still current.
/// A member added between the admin's `/pubkeys` read and this call would be
/// absent from the batch and dropped, so an admin racing itself loses a member
/// rather than leaking a key. Re-adding is one call; the alternative failure is
/// silent.
#[utoipa::path(post, path = "/api/v1/sync/groups/{id}/rotate", tag = "SyncKit",
    params(("id" = String, Path, description = "Group ID")),
    request_body = RotateGroupKeyRequest,
    responses(
        (status = 204, description = "Key rotated"),
        (status = 400, description = "Stale generation, or a grant set the server will not act on"),
        (status = 403, description = "Not the group admin"),
    ),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::rotate_group_key")]
pub(super) async fn rotate_key(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path(group_id): Path<SyncGroupId>,
    Json(req): Json<RotateGroupKeyRequest>,
) -> Result<impl IntoResponse> {
    let group = require_group(&db, sync_user.app_id, group_id).await?;
    require_admin(&db, group_id, sync_user.user_id).await?;

    // A generation that does not advance would re-point every member at a key
    // some previously-removed member may still hold.
    if req.gck_version <= group.gck_version {
        return Err(AppError::BadRequest(format!(
            "gck_version must be greater than the current generation ({})",
            group.gck_version
        )));
    }

    if req.grants.is_empty() {
        return Err(AppError::BadRequest(
            "A rotation must carry at least the admin's own grant".to_string(),
        ));
    }
    if req
        .grants
        .iter()
        .any(|g| g.sealed_gck.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES)
    {
        return Err(AppError::BadRequest(
            "Sealed key exceeds size limit".to_string(),
        ));
    }

    // The admin must be able to read the group afterwards. `rotate_group_gck`
    // deletes everyone outside the batch, so omitting the admin's own grant
    // orphans the group; the db layer states this as a precondition and nothing
    // enforced it.
    if !req.grants.iter().any(|g| g.user_id == group.admin_user_id) {
        return Err(AppError::BadRequest(
            "The rotation must include the admin's own re-sealed grant".to_string(),
        ));
    }

    // Rotation re-seals; it does not recruit. A grant for a non-member would be a
    // silent no-op in the db layer's UPDATE, so reject it here rather than let an
    // admin believe someone was added.
    let members: std::collections::HashSet<UserId> =
        db::synckit::list_member_pubkeys(&db, group_id)
            .await?
            .into_iter()
            .map(|(user_id, _)| user_id)
            .collect();
    let mut seen = std::collections::HashSet::with_capacity(req.grants.len());
    for grant in &req.grants {
        if !members.contains(&grant.user_id) {
            return Err(AppError::BadRequest(
                "A rotation grant names someone who is not a member; add members separately"
                    .to_string(),
            ));
        }
        if !seen.insert(grant.user_id) {
            return Err(AppError::BadRequest(
                "Duplicate grant for the same member".to_string(),
            ));
        }
    }

    let grants: Vec<(UserId, String)> = req
        .grants
        .into_iter()
        .map(|g| (g.user_id, g.sealed_gck))
        .collect();
    let removed = members.len().saturating_sub(grants.len());
    db::synckit::rotate_group_gck(&db, group_id, req.gck_version, &grants).await?;

    tracing::info!(
        %group_id,
        gck_version = req.gck_version,
        remaining = grants.len(),
        removed,
        "rotated group content key"
    );

    Ok(StatusCode::NO_CONTENT)
}

/// Remove a member from a group. Admin only.
///
/// Revocation only: it drops the member from future group writes but leaves the
/// GCK generation alone, so a member who kept a copy of the key can still read
/// any group ciphertext they can obtain. Forward secrecy comes from
/// [`rotate_key`], which removes and re-keys in one transaction and is what
/// `SyncKitClient::remove_member` drives. This endpoint remains for a caller that
/// wants revocation without a re-key. Data the member already pulled is already
/// in their hands either way.
#[utoipa::path(delete, path = "/api/v1/sync/groups/{id}/members/{user_id}", tag = "SyncKit",
    params(
        ("id" = String, Path, description = "Group ID"),
        ("user_id" = String, Path, description = "Member user ID"),
    ),
    responses((status = 204, description = "Member removed"), (status = 404, description = "Not a member")),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::remove_group_member")]
pub(super) async fn remove_member(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path((group_id, member_id)): Path<(SyncGroupId, UserId)>,
) -> Result<impl IntoResponse> {
    let group = require_group(&db, sync_user.app_id, group_id).await?;
    require_admin(&db, group_id, sync_user.user_id).await?;

    // The admin cannot remove themselves; that would orphan the group. Deleting a
    // group is a separate action (not yet exposed).
    if member_id == group.admin_user_id {
        return Err(AppError::BadRequest(
            "The group admin cannot be removed".to_string(),
        ));
    }

    if !db::synckit::remove_member(&db, group_id, member_id).await? {
        return Err(AppError::NotFound);
    }

    Ok(StatusCode::NO_CONTENT)
}

/// List a group's members (id, role, joined-at). Members only. Grants are not
/// included; each member fetches only their own via `/grant`.
#[utoipa::path(get, path = "/api/v1/sync/groups/{id}/members", tag = "SyncKit",
    params(("id" = String, Path, description = "Group ID")),
    responses((status = 200, description = "Group members", body = Vec<GroupMemberResponse>)),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::list_group_members")]
pub(super) async fn list_members(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path(group_id): Path<SyncGroupId>,
) -> Result<impl IntoResponse> {
    require_group(&db, sync_user.app_id, group_id).await?;
    require_member(&db, group_id, sync_user.user_id).await?;

    let members = db::synckit::list_members(&db, group_id).await?;
    let response: Vec<GroupMemberResponse> = members
        .into_iter()
        .map(|m| GroupMemberResponse {
            user_id: m.user_id,
            email: m.email,
            role: m.role,
            added_at: m.added_at,
        })
        .collect();
    Ok(Json(response))
}

/// Fetch the caller's own sealed GCK grant for a group, so their device can open
/// the GCK and read the group changelog. Members only.
///
/// Without `version`, returns the newest grant the caller holds, which is what a
/// device wants in order to write. With `version`, returns the grant for that
/// generation, which is what a device wants in order to read an entry pushed
/// before a rotation. A generation the caller never held is 403, the same answer
/// as not being a member: a non-member must not be able to probe which
/// generations exist.
#[utoipa::path(get, path = "/api/v1/sync/groups/{id}/grant", tag = "SyncKit",
    params(
        ("id" = String, Path, description = "Group ID"),
        ("version" = Option<i32>, Query, description = "GCK generation; omit for the newest"),
    ),
    responses(
        (status = 200, description = "The caller's sealed grant", body = GroupGrantResponse),
        (status = 403, description = "Not a member, or never held that generation"),
    ),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::get_group_grant")]
pub(super) async fn get_grant(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path(group_id): Path<SyncGroupId>,
    Query(query): Query<GrantQuery>,
) -> Result<impl IntoResponse> {
    require_group(&db, sync_user.app_id, group_id).await?;

    let (sealed_gck, gck_version) = match query.version {
        Some(version) => {
            let sealed =
                db::synckit::get_member_grant_at(&db, group_id, sync_user.user_id, version)
                    .await?
                    .ok_or(AppError::Forbidden)?;
            (sealed, version)
        }
        None => db::synckit::get_member_grant(&db, group_id, sync_user.user_id)
            .await?
            .ok_or(AppError::Forbidden)?,
    };

    Ok(Json(GroupGrantResponse {
        sealed_gck,
        gck_version,
    }))
}

/// Push encrypted changes to a group's shared changelog. Members only.
#[utoipa::path(post, path = "/api/v1/sync/groups/{id}/push", tag = "SyncKit",
    params(("id" = String, Path, description = "Group ID")),
    request_body = PushRequest,
    responses((status = 200, description = "New cursor position", body = PushResponse)),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::group_push", fields(group_id))]
pub(super) async fn group_push(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path(group_id): Path<SyncGroupId>,
    Json(req): Json<PushRequest>,
) -> Result<Response> {
    let group = require_group(&db, sync_user.app_id, group_id).await?;
    require_member(&db, group_id, sync_user.user_id).await?;

    // Group writes bill to the admin's slot (Groups billing decision): the paid
    // gate is checked against the *admin's* entitlement, not the pushing member's,
    // so a member with no subscription of their own can still contribute to a
    // group whose admin pays. Reads (group_pull) stay open, as personal pull does.
    if !db::synckit::internal_write_allowed(&db, sync_user.app_id, group.admin_user_id).await? {
        return Ok((
            StatusCode::PAYMENT_REQUIRED,
            Json(json!({ "reason": "no_subscription" })),
        )
            .into_response());
    }
    // NOTE: SSE push notification to group members is deferred to p3; members'
    // devices pick up group changes on their next scheduler tick meanwhile.

    if req.changes.is_empty() {
        return Err(AppError::BadRequest("No changes provided".to_string()));
    }
    if req.changes.len() > constants::SYNCKIT_PUSH_MAX_CHANGES {
        return Err(AppError::BadRequest(format!(
            "Maximum {} changes per push",
            constants::SYNCKIT_PUSH_MAX_CHANGES
        )));
    }
    for change in &req.changes {
        validation::validate_sync_table_name(&change.table)?;
        validation::validate_sync_row_id(&change.row_id)?;
        if change.op == db::SyncOperation::Delete && change.data.is_some() {
            return Err(AppError::BadRequest(
                "DELETE operations should not include data".to_string(),
            ));
        }
    }

    // The pushing device must belong to the pushing user (membership is a
    // separate, group-level check above).
    if !db::synckit::sync_device_belongs(&db, req.device_id, sync_user.app_id, sync_user.user_id)
        .await?
    {
        return Err(AppError::BadRequest("Unknown device".to_string()));
    }
    db::synckit::touch_sync_device(&db, req.device_id).await?;

    let changes: Vec<_> = req
        .changes
        .iter()
        .map(|c| {
            (
                c.table.clone(),
                c.op.to_string(),
                c.row_id.clone(),
                c.timestamp,
                c.data.clone(),
            )
        })
        .collect();

    let cursor = db::synckit::push_group_changes(
        &db,
        sync_user.app_id,
        group_id,
        sync_user.user_id,
        req.device_id,
        req.batch_id,
        &changes,
    )
    .await?;

    Ok(Json(PushResponse { cursor }).into_response())
}

/// Pull a group's changes after a cursor. Members only.
#[utoipa::path(post, path = "/api/v1/sync/groups/{id}/pull", tag = "SyncKit",
    params(("id" = String, Path, description = "Group ID")),
    request_body = PullRequest,
    responses((status = 200, description = "Changes since cursor", body = PullResponse)),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::group_pull", fields(group_id))]
pub(super) async fn group_pull(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path(group_id): Path<SyncGroupId>,
    Json(req): Json<PullRequest>,
) -> Result<impl IntoResponse> {
    require_group(&db, sync_user.app_id, group_id).await?;
    require_member(&db, group_id, sync_user.user_id).await?;

    if !db::synckit::sync_device_belongs(&db, req.device_id, sync_user.app_id, sync_user.user_id)
        .await?
    {
        return Err(AppError::BadRequest("Unknown device".to_string()));
    }

    if let Some(ref tables) = req.tables {
        if tables.len() > 50 {
            return Err(AppError::BadRequest(
                "Maximum 50 table names per filter".to_string(),
            ));
        }
        for table in tables {
            validation::validate_sync_table_name(table)?;
        }
    }

    let page_size = constants::SYNCKIT_PULL_PAGE_SIZE;
    let entries = db::synckit::pull_group_changes_filtered(
        &db,
        sync_user.app_id,
        group_id,
        req.cursor,
        page_size,
        req.tables.as_deref(),
        req.since,
    )
    .await?;

    let has_more = entries.len() as i64 == page_size;
    let new_cursor = entries.last().map_or(req.cursor, |e| e.seq);

    // Touch the device for activity, but do NOT advance the per-device personal
    // compaction cursor here; that cursor governs personal-changelog retention
    // and must not be moved by a group pull. Group changelog retention is a
    // separate concern (future work).
    db::synckit::touch_sync_device(&db, req.device_id).await?;

    let changes: Vec<PullChangeEntry> = entries
        .into_iter()
        .map(|e| PullChangeEntry {
            seq: e.seq,
            device_id: e.device_id,
            table: e.table_name,
            op: e.operation.to_string(),
            row_id: e.row_id,
            timestamp: e.client_timestamp,
            data: e.data,
            // Group entries key off the GCK generation stamped on the row, not a
            // per-user key_id.
            key_id: None,
            gck_version: Some(e.gck_version),
        })
        .collect();

    Ok(Json(PullResponse {
        changes,
        cursor: new_cursor,
        has_more,
    }))
}

// --- Invitations ---

/// Issue an invite link for a group. Admin only.
///
/// The token is minted here, hashed, and only the hash is stored, so this
/// response is the one and only time the plaintext exists server-side. The admin
/// puts it in a link and sends it however they like; the server is not involved
/// in delivery and never sees the link again until it is redeemed.
#[utoipa::path(post, path = "/api/v1/sync/groups/{id}/invitations", tag = "SyncKit",
    params(("id" = String, Path, description = "Group ID")),
    request_body = CreateInvitationRequest,
    responses(
        (status = 200, description = "Invitation issued", body = CreateInvitationResponse),
        (status = 403, description = "Not the group admin"),
    ),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::create_group_invitation")]
pub(super) async fn create_invitation(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path(group_id): Path<SyncGroupId>,
    Json(req): Json<CreateInvitationRequest>,
) -> Result<impl IntoResponse> {
    require_group(&db, sync_user.app_id, group_id).await?;
    require_admin(&db, group_id, sync_user.user_id).await?;

    let hours = req
        .expires_in_hours
        .unwrap_or(constants::SYNCKIT_INVITE_DEFAULT_HOURS);
    if !(constants::SYNCKIT_INVITE_MIN_HOURS..=constants::SYNCKIT_INVITE_MAX_HOURS).contains(&hours)
    {
        return Err(AppError::BadRequest(format!(
            "expires_in_hours must be between {} and {}",
            constants::SYNCKIT_INVITE_MIN_HOURS,
            constants::SYNCKIT_INVITE_MAX_HOURS
        )));
    }

    let token = generate_invite_token();
    let expires_at = Utc::now() + Duration::hours(hours);
    let invitation = db::synckit::create_invitation(
        &db,
        group_id,
        sync_user.user_id,
        &crate::crypto::sha256_hex(&token),
        expires_at,
    )
    .await?;

    tracing::info!(%group_id, invitation_id = %invitation.id, "issued group invite link");

    Ok(Json(CreateInvitationResponse {
        id: invitation.id,
        token,
        expires_at: invitation.expires_at,
    }))
}

/// A fresh invite token: 32 bytes of randomness, hex.
///
/// Hex rather than base64 so the token survives being pasted through anything
/// that mangles `+/=`, which a link handed between humans routinely is.
fn generate_invite_token() -> String {
    use rand::RngExt;
    let mut rng = rand::rng();
    let bytes: [u8; 32] = rng.random();
    hex::encode(bytes)
}

/// List a group's invitations. Admin only.
///
/// The admin's confirmation queue: an accepted invitation shows the invitee's
/// email and public key so the client can render a fingerprint to check against
/// what the invitee reads out over some other channel.
#[utoipa::path(get, path = "/api/v1/sync/groups/{id}/invitations", tag = "SyncKit",
    params(("id" = String, Path, description = "Group ID")),
    responses(
        (status = 200, description = "Invitations, newest first", body = Vec<InvitationResponse>),
        (status = 403, description = "Not the group admin"),
    ),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::list_group_invitations")]
pub(super) async fn list_invitations(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path(group_id): Path<SyncGroupId>,
) -> Result<impl IntoResponse> {
    require_group(&db, sync_user.app_id, group_id).await?;
    require_admin(&db, group_id, sync_user.user_id).await?;

    let invitations = db::synckit::list_invitations(&db, group_id).await?;
    let response: Vec<InvitationResponse> = invitations
        .into_iter()
        .map(InvitationResponse::from)
        .collect();
    Ok(Json(response))
}

/// Confirm an accepted invitation and seal the grant. Admin only.
///
/// This is the step the link deliberately does not remove. The admin has, by
/// this point, compared the invitee's key fingerprint against what the invitee
/// told them over a channel the server does not control; without that check a
/// server able to substitute a public key at acceptance would receive a grant to
/// the group key. Possession of a link gets someone into this queue and no
/// further.
///
/// The grant is sealed to the public key **stored on the invitation**, not to one
/// the admin re-supplies, so the key that was confirmed is the key that is used.
#[utoipa::path(post, path = "/api/v1/sync/groups/{id}/invitations/{invitation_id}/confirm", tag = "SyncKit",
    params(
        ("id" = String, Path, description = "Group ID"),
        ("invitation_id" = String, Path, description = "Invitation ID"),
    ),
    request_body = ConfirmInvitationRequest,
    responses(
        (status = 204, description = "Member added"),
        (status = 400, description = "Invitation is not awaiting confirmation"),
        (status = 403, description = "Not the group admin"),
    ),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::confirm_group_invitation")]
pub(super) async fn confirm_invitation(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path((group_id, invitation_id)): Path<(SyncGroupId, SyncGroupInvitationId)>,
    Json(req): Json<ConfirmInvitationRequest>,
) -> Result<impl IntoResponse> {
    let group = require_group(&db, sync_user.app_id, group_id).await?;
    require_admin(&db, group_id, sync_user.user_id).await?;

    if req.sealed_gck.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES {
        return Err(AppError::BadRequest(
            "Sealed key exceeds size limit".to_string(),
        ));
    }
    let role = req.role.as_deref().unwrap_or("member");
    if role != "member" && role != "admin" {
        return Err(AppError::BadRequest(
            "role must be 'member' or 'admin'".to_string(),
        ));
    }

    let invitation = db::synckit::get_invitation(&db, invitation_id)
        .await?
        .filter(|i| i.group_id == group_id)
        .ok_or(AppError::NotFound)?;

    if super::invitation_state(&invitation) != "accepted" {
        return Err(AppError::BadRequest(
            "That invitation is not awaiting confirmation".to_string(),
        ));
    }
    // An accepted invitation always carries both, but reading them out of Options
    // is where that invariant gets stated rather than assumed.
    let (Some(invitee_user_id), Some(invitee_pubkey)) = (
        invitation.invitee_user_id,
        invitation.invitee_pubkey.as_deref(),
    ) else {
        return Err(AppError::BadRequest(
            "That invitation has no accepted key".to_string(),
        ));
    };

    db::synckit::add_or_update_member(
        &db,
        group_id,
        invitee_user_id,
        role,
        &req.sealed_gck,
        group.gck_version,
        invitee_pubkey,
    )
    .await?;

    // Close the invitation after the grant lands. The other order would mark it
    // spent and then fail to add the member, leaving a token that opens nothing
    // and an invitee with no way back in but a fresh link.
    if !db::synckit::redeem_invitation(&db, group_id, invitation_id).await? {
        tracing::warn!(
            %group_id, %invitation_id,
            "member added but invitation was already closed; concurrent confirm"
        );
    }

    tracing::info!(%group_id, %invitation_id, "confirmed invitation and sealed grant");
    Ok(StatusCode::NO_CONTENT)
}

/// Cancel an invitation. Admin only.
///
/// Works on an accepted invitation as well as an outstanding one, because the
/// case that matters is an admin who looked at a fingerprint and did not
/// recognise it.
#[utoipa::path(delete, path = "/api/v1/sync/groups/{id}/invitations/{invitation_id}", tag = "SyncKit",
    params(
        ("id" = String, Path, description = "Group ID"),
        ("invitation_id" = String, Path, description = "Invitation ID"),
    ),
    responses(
        (status = 204, description = "Invitation revoked"),
        (status = 403, description = "Not the group admin"),
        (status = 404, description = "No such open invitation"),
    ),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::revoke_group_invitation")]
pub(super) async fn revoke_invitation(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Path((group_id, invitation_id)): Path<(SyncGroupId, SyncGroupInvitationId)>,
) -> Result<impl IntoResponse> {
    require_group(&db, sync_user.app_id, group_id).await?;
    require_admin(&db, group_id, sync_user.user_id).await?;

    if db::synckit::revoke_invitation(&db, group_id, invitation_id).await? {
        Ok(StatusCode::NO_CONTENT)
    } else {
        Err(AppError::NotFound)
    }
}

/// Show what an invite link leads to, before accepting it.
///
/// Authenticated, but not gated on membership: the caller is by definition not a
/// member yet. It answers which group, from whom, and whether the link is still
/// good. Nothing else is exposed, because anyone holding the link can read it.
#[utoipa::path(get, path = "/api/v1/sync/invitations/{token}", tag = "SyncKit",
    params(("token" = String, Path, description = "Invite token")),
    responses(
        (status = 200, description = "What the link leads to", body = InvitationPreviewResponse),
        (status = 404, description = "No such invitation"),
    ),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::preview_invitation")]
pub(super) async fn preview_invitation(
    State(db): State<PgPool>,
    _sync_user: SyncUser,
    Path(token): Path<String>,
) -> Result<impl IntoResponse> {
    let invitation = db::synckit::get_invitation_by_token(&db, &crate::crypto::sha256_hex(&token))
        .await?
        .ok_or(AppError::NotFound)?;

    // The group is read without an app scope because the token, not the caller,
    // established which group is meant; the token is unguessable and names
    // exactly one.
    let group = db::synckit::get_group_by_id(&db, invitation.group_id)
        .await?
        .ok_or(AppError::NotFound)?;
    let inviter_email = db::users::get_user_by_id(&db, invitation.inviter_user_id)
        .await?
        .map(|u| u.email.to_string())
        .unwrap_or_default();

    let state = super::invitation_state(&invitation);
    Ok(Json(InvitationPreviewResponse {
        group_name: group.name,
        inviter_email,
        redeemable: state == "pending",
        state: state.to_string(),
        expires_at: invitation.expires_at,
    }))
}

/// Accept an invitation by posting your identity public key against its token.
///
/// This grants nothing. It records the key the admin will seal the group key to
/// once they have confirmed its fingerprint, which is why an invitee is not a
/// member when this returns and their client should say so.
///
/// One-shot: the underlying update carries every liveness condition in its
/// predicate, so two devices racing the same link produce one acceptance.
#[utoipa::path(post, path = "/api/v1/sync/invitations/accept", tag = "SyncKit",
    request_body = AcceptInvitationRequest,
    responses(
        (status = 204, description = "Accepted; awaiting the admin's confirmation"),
        (status = 400, description = "Link is expired, revoked, or already used"),
        (status = 409, description = "Already a member, or already awaiting confirmation"),
    ),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::accept_invitation")]
pub(super) async fn accept_invitation(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    Json(req): Json<AcceptInvitationRequest>,
) -> Result<impl IntoResponse> {
    if req.invitee_pubkey.is_empty()
        || req.invitee_pubkey.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES
    {
        return Err(AppError::BadRequest("Invalid public key".to_string()));
    }

    let token_hash = crate::crypto::sha256_hex(&req.token);

    // Read first, only to tell the invitee why a link that plainly exists does
    // not work. The read is not the gate; the conditional update below is, so a
    // token going stale between the two changes the message and not the outcome.
    let existing = db::synckit::get_invitation_by_token(&db, &token_hash)
        .await?
        .ok_or(AppError::NotFound)?;

    if db::synckit::is_group_member(&db, existing.group_id, sync_user.user_id).await? {
        return Err(AppError::Conflict(
            "You are already a member of that group".to_string(),
        ));
    }

    let accepted =
        db::synckit::accept_invitation(&db, &token_hash, sync_user.user_id, &req.invitee_pubkey)
            .await?;

    match accepted {
        Some(invitation) => {
            tracing::info!(
                group_id = %invitation.group_id,
                invitation_id = %invitation.id,
                "invitation accepted; awaiting admin confirmation"
            );
            Ok(StatusCode::NO_CONTENT)
        }
        None => Err(AppError::BadRequest(format!(
            "That invite link is {}",
            super::invitation_state(&existing)
        ))),
    }
}