//! PostgreSQL implementation of [`crate::repo::traits::ChatRepository`].
//!
//! Maps domain rows to [`crate::dto`] views, enforces [`crate::authz`] checks, and writes outbox
//! events for [`crate::ChatApp`] publication.

use std::collections::HashSet;

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde_json::json;
use sqlx::{PgPool, Postgres, Row, Transaction};
use uuid::Uuid;

use crate::authz::ensure;
use crate::dto::commands::{
    AddMembers, AttachmentInput, CreateRoom, EditMessage, ListMessages, ListRooms, SearchMessages,
    SendMessage, UpdateRoom,
};
use crate::dto::results::{
    AttachmentView, MemberView, MessagePage, MessageView, ReactionCount, ReactionSummary, RoomPage,
    RoomView, SearchHit, SearchPage, annotate_message_groups,
};
use crate::error::{ChatError, ChatResult};
use crate::events::{ChatEventEnvelope, ChatEventPayload};
use crate::ids::{MessageId, RoomId, UserId};
use crate::outbox::insert_chat_outbox_tx;
use crate::repo::traits::{
    ChatRepository, DeleteMessageCommit, MembersCommit, MessageMutationCommit, ReactionCommit,
    ReadCursorCommit, RoomCommit, SendMessageCommit,
};

/// Stateless SQLx-backed [`ChatRepository`] used by [`crate::SqlxChatApp`].
#[derive(Debug, Default)]
pub struct SqlxChatRepository;

impl SqlxChatRepository {
    /// Returns the default repository instance.
    pub fn new() -> Self {
        Self
    }
}

struct RoomAccess {
    organization_id: String,
    role: String,
    version: i64,
    last_message_seq: i64,
}

struct MessageGroupBoundary {
    sender_id: UserId,
    created_at: DateTime<Utc>,
}

fn room_view_from_row(row: &sqlx::postgres::PgRow) -> ChatResult<RoomView> {
    Ok(RoomView {
        id: row.try_get("id")?,
        organization_id: row.try_get("organization_id")?,
        kind: row.try_get("kind")?,
        title: row.try_get("title")?,
        created_by: row.try_get("created_by")?,
        created_at: row.try_get("created_at")?,
        updated_at: row.try_get("updated_at")?,
        archived_at: row.try_get("archived_at")?,
        last_message_id: row.try_get("last_message_id")?,
        last_message_seq: row.try_get("last_message_seq")?,
        last_message_at: row.try_get("last_message_at")?,
        version: row.try_get("version")?,
    })
}

fn member_view_from_row(row: &sqlx::postgres::PgRow) -> ChatResult<MemberView> {
    Ok(MemberView {
        room_id: row.try_get("room_id")?,
        user_id: row.try_get("user_id")?,
        role: row.try_get("role")?,
        joined_at: row.try_get("joined_at")?,
        last_read_seq: row.try_get("last_read_seq")?,
        last_read_message_id: row.try_get("last_read_message_id")?,
        muted: row.try_get("muted")?,
        notification_mode: row.try_get("notification_mode")?,
        hidden_at: row.try_get("hidden_at")?,
    })
}

async fn load_room_access(
    executor: &mut Transaction<'_, Postgres>,
    room_id: RoomId,
    organization_id: &str,
    actor: &str,
) -> ChatResult<RoomAccess> {
    let row = sqlx::query(
        r#"
        SELECT
            r.organization_id,
            r.version,
            r.last_message_seq,
            m.role
        FROM athena.chat_rooms AS r
        INNER JOIN athena.chat_room_members AS m
            ON m.room_id = r.id
        WHERE r.id = $1
          AND r.organization_id = $2
          AND m.user_id = $3
          AND m.hidden_at IS NULL
        FOR UPDATE OF r
        "#,
    )
    .bind(room_id)
    .bind(organization_id)
    .bind(actor)
    .fetch_optional(&mut **executor)
    .await?;

    let Some(row) = row else {
        return Err(ChatError::NotFound(
            "room not found or inaccessible".to_string(),
        ));
    };

    Ok(RoomAccess {
        organization_id: row.try_get("organization_id")?,
        role: row.try_get("role")?,
        version: row.try_get("version")?,
        last_message_seq: row.try_get("last_message_seq")?,
    })
}

async fn load_room_access_read(
    pool: &PgPool,
    room_id: RoomId,
    organization_id: &str,
    actor: &str,
) -> ChatResult<RoomAccess> {
    let row = sqlx::query(
        r#"
        SELECT
            r.organization_id,
            r.version,
            r.last_message_seq,
            m.role
        FROM athena.chat_rooms AS r
        INNER JOIN athena.chat_room_members AS m
            ON m.room_id = r.id
        WHERE r.id = $1
          AND r.organization_id = $2
          AND m.user_id = $3
          AND m.hidden_at IS NULL
        "#,
    )
    .bind(room_id)
    .bind(organization_id)
    .bind(actor)
    .fetch_optional(pool)
    .await?;

    let Some(row) = row else {
        return Err(ChatError::NotFound(
            "room not found or inaccessible".to_string(),
        ));
    };

    Ok(RoomAccess {
        organization_id: row.try_get("organization_id")?,
        role: row.try_get("role")?,
        version: row.try_get("version")?,
        last_message_seq: row.try_get("last_message_seq")?,
    })
}

fn actor_can_manage_members(role: &str) -> bool {
    matches!(role, "owner" | "admin")
}

async fn load_room_view(pool: &PgPool, room_id: RoomId) -> ChatResult<RoomView> {
    let row = sqlx::query("SELECT * FROM athena.chat_rooms WHERE id = $1")
        .bind(room_id)
        .fetch_one(pool)
        .await?;
    room_view_from_row(&row)
}

async fn load_room_view_tx(
    tx: &mut Transaction<'_, Postgres>,
    room_id: RoomId,
) -> ChatResult<RoomView> {
    let row = sqlx::query("SELECT * FROM athena.chat_rooms WHERE id = $1")
        .bind(room_id)
        .fetch_one(&mut **tx)
        .await?;
    room_view_from_row(&row)
}

async fn load_message_attachments(
    pool: &PgPool,
    message_id: MessageId,
) -> ChatResult<Vec<AttachmentView>> {
    let rows = sqlx::query(
        r#"
        SELECT
            a.file_id,
            a.ordinal,
            f.file_name,
            f.original_name,
            f.content_type,
            f.mime_type,
            f.extension,
            f.size_bytes,
            f.status,
            f.visibility,
            f.url AS file_url,
            f.bucket,
            f.storage_key
        FROM athena.chat_message_attachments AS a
        LEFT JOIN athena.files AS f
            ON f.id = a.file_id
        WHERE a.message_id = $1
        ORDER BY a.ordinal ASC
        "#,
    )
    .bind(message_id)
    .fetch_all(pool)
    .await?;

    let mut items = Vec::with_capacity(rows.len());
    for row in rows {
        let file_id: String = row.try_get("file_id")?;
        items.push(AttachmentView {
            file_id: file_id.clone(),
            ordinal: row.try_get("ordinal")?,
            file_name: row.try_get("file_name")?,
            original_name: row.try_get("original_name")?,
            content_type: row.try_get("content_type")?,
            mime_type: row.try_get("mime_type")?,
            extension: row.try_get("extension")?,
            size_bytes: row.try_get("size_bytes")?,
            status: row.try_get("status")?,
            visibility: row.try_get("visibility")?,
            file_url: row.try_get("file_url")?,
            bucket: row.try_get("bucket")?,
            storage_key: row.try_get("storage_key")?,
            authorized_url_path: format!("/storage/files/{file_id}/url"),
            proxy_url_path: format!("/storage/files/{file_id}/proxy"),
            public_url_path: format!("/storage/files/{file_id}/public-url"),
        });
    }
    Ok(items)
}

async fn load_message_attachments_tx(
    tx: &mut Transaction<'_, Postgres>,
    message_id: MessageId,
) -> ChatResult<Vec<AttachmentView>> {
    let rows = sqlx::query(
        r#"
        SELECT
            a.file_id,
            a.ordinal,
            f.file_name,
            f.original_name,
            f.content_type,
            f.mime_type,
            f.extension,
            f.size_bytes,
            f.status,
            f.visibility,
            f.url AS file_url,
            f.bucket,
            f.storage_key
        FROM athena.chat_message_attachments AS a
        LEFT JOIN athena.files AS f
            ON f.id = a.file_id
        WHERE a.message_id = $1
        ORDER BY a.ordinal ASC
        "#,
    )
    .bind(message_id)
    .fetch_all(&mut **tx)
    .await?;

    let mut items = Vec::with_capacity(rows.len());
    for row in rows {
        let file_id: String = row.try_get("file_id")?;
        items.push(AttachmentView {
            file_id: file_id.clone(),
            ordinal: row.try_get("ordinal")?,
            file_name: row.try_get("file_name")?,
            original_name: row.try_get("original_name")?,
            content_type: row.try_get("content_type")?,
            mime_type: row.try_get("mime_type")?,
            extension: row.try_get("extension")?,
            size_bytes: row.try_get("size_bytes")?,
            status: row.try_get("status")?,
            visibility: row.try_get("visibility")?,
            file_url: row.try_get("file_url")?,
            bucket: row.try_get("bucket")?,
            storage_key: row.try_get("storage_key")?,
            authorized_url_path: format!("/storage/files/{file_id}/url"),
            proxy_url_path: format!("/storage/files/{file_id}/proxy"),
            public_url_path: format!("/storage/files/{file_id}/public-url"),
        });
    }
    Ok(items)
}

async fn load_reaction_summary(
    pool: &PgPool,
    message_id: MessageId,
    actor: &str,
) -> ChatResult<ReactionSummary> {
    let rows = sqlx::query(
        r#"
        SELECT
            emoji,
            COUNT(*)::bigint AS reaction_count,
            BOOL_OR(user_id = $2) AS reacted
        FROM athena.chat_message_reactions
        WHERE message_id = $1
        GROUP BY emoji
        ORDER BY emoji ASC
        "#,
    )
    .bind(message_id)
    .bind(actor)
    .fetch_all(pool)
    .await?;

    let mut reactions = Vec::with_capacity(rows.len());
    for row in rows {
        reactions.push(ReactionCount {
            emoji: row.try_get("emoji")?,
            count: row.try_get("reaction_count")?,
            reacted: row.try_get("reacted")?,
        });
    }

    Ok(ReactionSummary {
        message_id,
        reactions,
    })
}

async fn load_reaction_summary_tx(
    tx: &mut Transaction<'_, Postgres>,
    message_id: MessageId,
    actor: &str,
) -> ChatResult<ReactionSummary> {
    let rows = sqlx::query(
        r#"
        SELECT
            emoji,
            COUNT(*)::bigint AS reaction_count,
            BOOL_OR(user_id = $2) AS reacted
        FROM athena.chat_message_reactions
        WHERE message_id = $1
        GROUP BY emoji
        ORDER BY emoji ASC
        "#,
    )
    .bind(message_id)
    .bind(actor)
    .fetch_all(&mut **tx)
    .await?;

    let mut reactions = Vec::with_capacity(rows.len());
    for row in rows {
        reactions.push(ReactionCount {
            emoji: row.try_get("emoji")?,
            count: row.try_get("reaction_count")?,
            reacted: row.try_get("reacted")?,
        });
    }

    Ok(ReactionSummary {
        message_id,
        reactions,
    })
}

async fn load_message_view(
    pool: &PgPool,
    message_id: MessageId,
    actor: &str,
) -> ChatResult<MessageView> {
    let mut view = load_message_view_base(pool, message_id, actor).await?;
    let previous_boundary =
        load_previous_message_boundary(pool, view.room_id, view.room_seq).await?;
    let next_boundary = load_next_message_boundary(pool, view.room_id, view.room_seq).await?;
    annotate_message_groups(
        std::slice::from_mut(&mut view),
        previous_boundary
            .as_ref()
            .map(|boundary| (boundary.sender_id.as_str(), boundary.created_at)),
        next_boundary
            .as_ref()
            .map(|boundary| (boundary.sender_id.as_str(), boundary.created_at)),
    );
    Ok(view)
}

async fn load_message_view_base(
    pool: &PgPool,
    message_id: MessageId,
    actor: &str,
) -> ChatResult<MessageView> {
    let row = sqlx::query(
        r#"
        SELECT
            id,
            room_id,
            room_seq,
            sender_id,
            client_message_id,
            body_text,
            body_json,
            reply_to_message_id,
            created_at,
            edited_at,
            deleted_at,
            metadata_json
        FROM athena.chat_messages
        WHERE id = $1
        "#,
    )
    .bind(message_id)
    .fetch_one(pool)
    .await?;

    let attachments = load_message_attachments(pool, message_id).await?;
    let reactions = load_reaction_summary(pool, message_id, actor)
        .await?
        .reactions;

    Ok(MessageView {
        id: row.try_get("id")?,
        room_id: row.try_get("room_id")?,
        room_seq: row.try_get("room_seq")?,
        sender_id: row.try_get("sender_id")?,
        client_message_id: row.try_get("client_message_id")?,
        body_text: row.try_get("body_text")?,
        body_json: row.try_get("body_json")?,
        reply_to_message_id: row.try_get("reply_to_message_id")?,
        created_at: row.try_get("created_at")?,
        edited_at: row.try_get("edited_at")?,
        deleted_at: row.try_get("deleted_at")?,
        metadata_json: row.try_get("metadata_json")?,
        attachments,
        reactions,
        message_group: None,
    })
}

async fn load_message_view_tx(
    tx: &mut Transaction<'_, Postgres>,
    message_id: MessageId,
    actor: &str,
) -> ChatResult<MessageView> {
    let mut view = load_message_view_base_tx(tx, message_id, actor).await?;
    let previous_boundary =
        load_previous_message_boundary_tx(tx, view.room_id, view.room_seq).await?;
    let next_boundary = load_next_message_boundary_tx(tx, view.room_id, view.room_seq).await?;
    annotate_message_groups(
        std::slice::from_mut(&mut view),
        previous_boundary
            .as_ref()
            .map(|boundary| (boundary.sender_id.as_str(), boundary.created_at)),
        next_boundary
            .as_ref()
            .map(|boundary| (boundary.sender_id.as_str(), boundary.created_at)),
    );
    Ok(view)
}

async fn load_message_view_base_tx(
    tx: &mut Transaction<'_, Postgres>,
    message_id: MessageId,
    actor: &str,
) -> ChatResult<MessageView> {
    let row = sqlx::query(
        r#"
        SELECT
            id,
            room_id,
            room_seq,
            sender_id,
            client_message_id,
            body_text,
            body_json,
            reply_to_message_id,
            created_at,
            edited_at,
            deleted_at,
            metadata_json
        FROM athena.chat_messages
        WHERE id = $1
        "#,
    )
    .bind(message_id)
    .fetch_one(&mut **tx)
    .await?;

    let attachments = load_message_attachments_tx(tx, message_id).await?;
    let reactions = load_reaction_summary_tx(tx, message_id, actor)
        .await?
        .reactions;

    Ok(MessageView {
        id: row.try_get("id")?,
        room_id: row.try_get("room_id")?,
        room_seq: row.try_get("room_seq")?,
        sender_id: row.try_get("sender_id")?,
        client_message_id: row.try_get("client_message_id")?,
        body_text: row.try_get("body_text")?,
        body_json: row.try_get("body_json")?,
        reply_to_message_id: row.try_get("reply_to_message_id")?,
        created_at: row.try_get("created_at")?,
        edited_at: row.try_get("edited_at")?,
        deleted_at: row.try_get("deleted_at")?,
        metadata_json: row.try_get("metadata_json")?,
        attachments,
        reactions,
        message_group: None,
    })
}

async fn load_previous_message_boundary(
    pool: &PgPool,
    room_id: RoomId,
    room_seq: i64,
) -> ChatResult<Option<MessageGroupBoundary>> {
    let row = sqlx::query(
        r#"
        SELECT sender_id, created_at
        FROM athena.chat_messages
        WHERE room_id = $1
          AND room_seq < $2
        ORDER BY room_seq DESC
        LIMIT 1
        "#,
    )
    .bind(room_id)
    .bind(room_seq)
    .fetch_optional(pool)
    .await?;

    row.map(|row| {
        Ok(MessageGroupBoundary {
            sender_id: row.try_get("sender_id")?,
            created_at: row.try_get("created_at")?,
        })
    })
    .transpose()
}

async fn load_next_message_boundary(
    pool: &PgPool,
    room_id: RoomId,
    room_seq: i64,
) -> ChatResult<Option<MessageGroupBoundary>> {
    let row = sqlx::query(
        r#"
        SELECT sender_id, created_at
        FROM athena.chat_messages
        WHERE room_id = $1
          AND room_seq > $2
        ORDER BY room_seq ASC
        LIMIT 1
        "#,
    )
    .bind(room_id)
    .bind(room_seq)
    .fetch_optional(pool)
    .await?;

    row.map(|row| {
        Ok(MessageGroupBoundary {
            sender_id: row.try_get("sender_id")?,
            created_at: row.try_get("created_at")?,
        })
    })
    .transpose()
}

async fn load_previous_message_boundary_tx(
    tx: &mut Transaction<'_, Postgres>,
    room_id: RoomId,
    room_seq: i64,
) -> ChatResult<Option<MessageGroupBoundary>> {
    let row = sqlx::query(
        r#"
        SELECT sender_id, created_at
        FROM athena.chat_messages
        WHERE room_id = $1
          AND room_seq < $2
        ORDER BY room_seq DESC
        LIMIT 1
        "#,
    )
    .bind(room_id)
    .bind(room_seq)
    .fetch_optional(&mut **tx)
    .await?;

    row.map(|row| {
        Ok(MessageGroupBoundary {
            sender_id: row.try_get("sender_id")?,
            created_at: row.try_get("created_at")?,
        })
    })
    .transpose()
}

async fn load_next_message_boundary_tx(
    tx: &mut Transaction<'_, Postgres>,
    room_id: RoomId,
    room_seq: i64,
) -> ChatResult<Option<MessageGroupBoundary>> {
    let row = sqlx::query(
        r#"
        SELECT sender_id, created_at
        FROM athena.chat_messages
        WHERE room_id = $1
          AND room_seq > $2
        ORDER BY room_seq ASC
        LIMIT 1
        "#,
    )
    .bind(room_id)
    .bind(room_seq)
    .fetch_optional(&mut **tx)
    .await?;

    row.map(|row| {
        Ok(MessageGroupBoundary {
            sender_id: row.try_get("sender_id")?,
            created_at: row.try_get("created_at")?,
        })
    })
    .transpose()
}

async fn insert_message_attachments(
    tx: &mut Transaction<'_, Postgres>,
    message_id: MessageId,
    organization_id: &str,
    attachments: &[AttachmentInput],
) -> ChatResult<()> {
    for attachment in attachments {
        let result = sqlx::query(
            r#"
            INSERT INTO athena.chat_message_attachments
                (message_id, file_id, ordinal)
            SELECT
                $1,
                f.id,
                $3
            FROM athena.files AS f
            WHERE f.id = $2
              AND f.organization_id = $4
              AND f.deleted_at IS NULL
            "#,
        )
        .bind(message_id)
        .bind(&attachment.file_id)
        .bind(attachment.ordinal)
        .bind(organization_id)
        .execute(&mut **tx)
        .await?;
        if result.rows_affected() != 1 {
            return Err(ChatError::Validation(format!(
                "attachment file '{}' is not available for this organization",
                attachment.file_id
            )));
        }
    }
    Ok(())
}

fn message_event(room_id: RoomId, room_seq: i64, payload: ChatEventPayload) -> ChatEventEnvelope {
    ChatEventEnvelope {
        event_id: Uuid::new_v4(),
        room_id,
        room_seq: Some(room_seq),
        occurred_at: Utc::now(),
        payload,
    }
}

fn room_event(room_id: RoomId, payload: ChatEventPayload) -> ChatEventEnvelope {
    ChatEventEnvelope {
        event_id: Uuid::new_v4(),
        room_id,
        room_seq: None,
        occurred_at: Utc::now(),
        payload,
    }
}

#[async_trait]
impl ChatRepository for SqlxChatRepository {
    async fn create_room(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        input: CreateRoom,
    ) -> ChatResult<RoomCommit> {
        let mut tx = pool.begin().await?;
        let room_id = Uuid::new_v4();
        sqlx::query(
            r#"
            INSERT INTO athena.chat_rooms
                (id, organization_id, kind, title, created_by, last_message_seq, version)
            VALUES
                ($1, $2, $3, $4, $5, 0, 1)
            "#,
        )
        .bind(room_id)
        .bind(organization_id)
        .bind(&input.kind)
        .bind(&input.title)
        .bind(actor)
        .execute(&mut *tx)
        .await?;

        let mut members = HashSet::new();
        members.insert(actor.to_string());
        for user_id in input.member_user_ids {
            if !user_id.trim().is_empty() {
                members.insert(user_id);
            }
        }

        for user_id in members {
            let role = if user_id == actor { "owner" } else { "member" };
            sqlx::query(
                r#"
                INSERT INTO athena.chat_room_members
                    (room_id, user_id, role, last_read_seq, muted)
                VALUES
                    ($1, $2, $3, 0, false)
                ON CONFLICT (room_id, user_id) DO NOTHING
                "#,
            )
            .bind(room_id)
            .bind(user_id)
            .bind(role)
            .execute(&mut *tx)
            .await?;
        }

        let room = load_room_view_tx(&mut tx, room_id).await?;
        let event = room_event(
            room_id,
            ChatEventPayload::RoomCreated { room: room.clone() },
        );
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_room",
            &room.id.to_string(),
            room.id,
            None,
            "chat.room.created",
            &event,
        )
        .await?;
        tx.commit().await?;
        Ok(RoomCommit {
            room,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn update_room(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        input: UpdateRoom,
    ) -> ChatResult<RoomCommit> {
        let mut tx = pool.begin().await?;
        let access = load_room_access(&mut tx, input.room_id, organization_id, actor).await?;
        ensure(
            actor_can_manage_members(&access.role),
            "room update requires admin or owner role",
        )?;

        sqlx::query(
            r#"
            UPDATE athena.chat_rooms
            SET title = $2,
                updated_at = now(),
                version = version + 1
            WHERE id = $1
            "#,
        )
        .bind(input.room_id)
        .bind(&input.title)
        .execute(&mut *tx)
        .await?;

        let room = load_room_view_tx(&mut tx, input.room_id).await?;
        let event = room_event(
            input.room_id,
            ChatEventPayload::RoomUpdated { room: room.clone() },
        );
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_room",
            &room.id.to_string(),
            room.id,
            None,
            "chat.room.updated",
            &event,
        )
        .await?;
        tx.commit().await?;
        Ok(RoomCommit {
            room,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn archive_room(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        room_id: RoomId,
    ) -> ChatResult<RoomCommit> {
        let mut tx = pool.begin().await?;
        let access = load_room_access(&mut tx, room_id, organization_id, actor).await?;
        ensure(
            actor_can_manage_members(&access.role),
            "room archive requires admin or owner role",
        )?;

        sqlx::query(
            r#"
            UPDATE athena.chat_rooms
            SET archived_at = now(),
                updated_at = now(),
                version = version + 1
            WHERE id = $1
            "#,
        )
        .bind(room_id)
        .execute(&mut *tx)
        .await?;
        let room = load_room_view_tx(&mut tx, room_id).await?;
        let event = room_event(
            room_id,
            ChatEventPayload::RoomArchived { room: room.clone() },
        );
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_room",
            &room.id.to_string(),
            room.id,
            None,
            "chat.room.archived",
            &event,
        )
        .await?;
        tx.commit().await?;
        Ok(RoomCommit {
            room,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn list_rooms(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        input: ListRooms,
    ) -> ChatResult<RoomPage> {
        let limit = input.limit.unwrap_or(50).clamp(1, 200);
        let offset = input.offset.unwrap_or(0).max(0);
        let include_archived = input.include_archived.unwrap_or(false);
        let rows = sqlx::query(
            r#"
            SELECT r.*
            FROM athena.chat_rooms AS r
            INNER JOIN athena.chat_room_members AS m
                ON m.room_id = r.id
            WHERE r.organization_id = $1
              AND m.user_id = $2
              AND m.hidden_at IS NULL
              AND ($3::bool OR r.archived_at IS NULL)
            ORDER BY COALESCE(r.last_message_at, r.created_at) DESC, r.updated_at DESC
            LIMIT $4 OFFSET $5
            "#,
        )
        .bind(organization_id)
        .bind(actor)
        .bind(include_archived)
        .bind(limit)
        .bind(offset)
        .fetch_all(pool)
        .await?;

        let mut items = Vec::with_capacity(rows.len());
        for row in rows {
            items.push(room_view_from_row(&row)?);
        }
        Ok(RoomPage { items })
    }

    async fn get_room(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        room_id: RoomId,
    ) -> ChatResult<RoomView> {
        let _ = load_room_access_read(pool, room_id, organization_id, actor).await?;
        load_room_view(pool, room_id).await
    }

    async fn list_members(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        room_id: RoomId,
    ) -> ChatResult<Vec<MemberView>> {
        let _ = load_room_access_read(pool, room_id, organization_id, actor).await?;
        let rows = sqlx::query(
            "SELECT * FROM athena.chat_room_members WHERE room_id = $1 ORDER BY joined_at ASC",
        )
        .bind(room_id)
        .fetch_all(pool)
        .await?;
        let mut items = Vec::with_capacity(rows.len());
        for row in rows {
            items.push(member_view_from_row(&row)?);
        }
        Ok(items)
    }

    async fn add_members(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        input: AddMembers,
    ) -> ChatResult<MembersCommit> {
        let mut tx = pool.begin().await?;
        let access = load_room_access(&mut tx, input.room_id, organization_id, actor).await?;
        ensure(
            actor_can_manage_members(&access.role),
            "member add requires admin or owner role",
        )?;

        let role = input.role.unwrap_or_else(|| "member".to_string());
        for user_id in input.user_ids {
            sqlx::query(
                r#"
                INSERT INTO athena.chat_room_members
                    (room_id, user_id, role, last_read_seq, muted)
                VALUES
                    ($1, $2, $3, 0, false)
                ON CONFLICT (room_id, user_id) DO UPDATE
                SET role = EXCLUDED.role,
                    hidden_at = NULL
                "#,
            )
            .bind(input.room_id)
            .bind(user_id)
            .bind(&role)
            .execute(&mut *tx)
            .await?;
        }

        sqlx::query(
            "UPDATE athena.chat_rooms SET version = version + 1, updated_at = now() WHERE id = $1",
        )
        .bind(input.room_id)
        .execute(&mut *tx)
        .await?;

        let room = sqlx::query("SELECT version FROM athena.chat_rooms WHERE id = $1")
            .bind(input.room_id)
            .fetch_one(&mut *tx)
            .await?;
        let version: i64 = room.try_get("version")?;
        let event = room_event(input.room_id, ChatEventPayload::MembersUpdated { version });
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_room",
            &input.room_id.to_string(),
            input.room_id,
            None,
            "chat.members.updated",
            &event,
        )
        .await?;
        tx.commit().await?;
        let members = self
            .list_members(pool, organization_id, actor, input.room_id)
            .await?;
        Ok(MembersCommit {
            members,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn remove_member(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        room_id: RoomId,
        user_id: &str,
    ) -> ChatResult<MembersCommit> {
        let mut tx = pool.begin().await?;
        let access = load_room_access(&mut tx, room_id, organization_id, actor).await?;
        ensure(
            actor_can_manage_members(&access.role) || actor == user_id,
            "member removal requires admin/owner or self-removal",
        )?;

        sqlx::query("DELETE FROM athena.chat_room_members WHERE room_id = $1 AND user_id = $2")
            .bind(room_id)
            .bind(user_id)
            .execute(&mut *tx)
            .await?;

        sqlx::query(
            "UPDATE athena.chat_rooms SET version = version + 1, updated_at = now() WHERE id = $1",
        )
        .bind(room_id)
        .execute(&mut *tx)
        .await?;
        let room = sqlx::query("SELECT version FROM athena.chat_rooms WHERE id = $1")
            .bind(room_id)
            .fetch_one(&mut *tx)
            .await?;
        let version: i64 = room.try_get("version")?;
        let event = room_event(room_id, ChatEventPayload::MembersUpdated { version });
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_room",
            &room_id.to_string(),
            room_id,
            None,
            "chat.members.updated",
            &event,
        )
        .await?;
        tx.commit().await?;
        let members = self
            .list_members(pool, organization_id, actor, room_id)
            .await
            .unwrap_or_default();
        Ok(MembersCommit {
            members,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn list_messages(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        input: ListMessages,
    ) -> ChatResult<MessagePage> {
        let _ = load_room_access_read(pool, input.room_id, organization_id, actor).await?;
        let limit = input.limit.unwrap_or(50).clamp(1, 200);
        let rows = sqlx::query(
            r#"
            SELECT
                id,
                room_id,
                room_seq,
                sender_id,
                client_message_id,
                body_text,
                body_json,
                reply_to_message_id,
                created_at,
                edited_at,
                deleted_at,
                metadata_json
            FROM athena.chat_messages
            WHERE room_id = $1
              AND ($2::bigint IS NULL OR room_seq < $2)
              AND ($3::bigint IS NULL OR room_seq > $3)
            ORDER BY room_seq DESC
            LIMIT $4
            "#,
        )
        .bind(input.room_id)
        .bind(input.before_seq)
        .bind(input.after_seq)
        .bind(limit)
        .fetch_all(pool)
        .await?;

        let mut items = Vec::with_capacity(rows.len());
        let mut next_before_seq = None;
        for row in rows {
            let id: MessageId = row.try_get("id")?;
            let view = load_message_view_base(pool, id, actor).await?;
            next_before_seq = Some(view.room_seq);
            items.push(view);
        }
        items.reverse();
        if let (Some(first), Some(last)) = (items.first(), items.last()) {
            let previous_boundary =
                load_previous_message_boundary(pool, first.room_id, first.room_seq).await?;
            let next_boundary =
                load_next_message_boundary(pool, last.room_id, last.room_seq).await?;
            annotate_message_groups(
                &mut items,
                previous_boundary
                    .as_ref()
                    .map(|boundary| (boundary.sender_id.as_str(), boundary.created_at)),
                next_boundary
                    .as_ref()
                    .map(|boundary| (boundary.sender_id.as_str(), boundary.created_at)),
            );
        }
        Ok(MessagePage {
            items,
            next_before_seq,
        })
    }

    async fn send_message(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        input: SendMessage,
    ) -> ChatResult<SendMessageCommit> {
        let mut tx = pool.begin().await?;
        let access = load_room_access(&mut tx, input.room_id, organization_id, actor).await?;

        if let Some(client_message_id) = input.client_message_id.as_ref() {
            if let Some(existing) = sqlx::query(
                r#"
                SELECT id
                FROM athena.chat_messages
                WHERE room_id = $1
                  AND sender_id = $2
                  AND client_message_id = $3
                "#,
            )
            .bind(input.room_id)
            .bind(actor)
            .bind(client_message_id)
            .fetch_optional(&mut *tx)
            .await?
            {
                let message_id: MessageId = existing.try_get("id")?;
                drop(tx);
                let message = load_message_view(pool, message_id, actor).await?;
                let event = message_event(
                    input.room_id,
                    message.room_seq,
                    ChatEventPayload::MessageCreated {
                        message: message.clone(),
                    },
                );
                return Ok(SendMessageCommit {
                    message,
                    event,
                    outbox_id: None,
                });
            }
        }

        let next_seq = access.last_message_seq + 1;
        let message_id = Uuid::new_v4();
        sqlx::query(
            r#"
            INSERT INTO athena.chat_messages
                (id, room_id, room_seq, sender_id, client_message_id, body_text, body_json, reply_to_message_id, metadata_json)
            VALUES
                ($1, $2, $3, $4, $5, $6, $7, $8, $9)
            "#,
        )
        .bind(message_id)
        .bind(input.room_id)
        .bind(next_seq)
        .bind(actor)
        .bind(&input.client_message_id)
        .bind(&input.body_text)
        .bind(&input.body_json)
        .bind(input.reply_to_message_id)
        .bind(&input.metadata_json)
        .execute(&mut *tx)
        .await?;

        insert_message_attachments(&mut tx, message_id, organization_id, &input.attachments)
            .await?;

        sqlx::query(
            r#"
            UPDATE athena.chat_rooms
            SET last_message_id = $2,
                last_message_seq = $3,
                last_message_at = now(),
                updated_at = now()
            WHERE id = $1
            "#,
        )
        .bind(input.room_id)
        .bind(message_id)
        .bind(next_seq)
        .execute(&mut *tx)
        .await?;

        let message = load_message_view_tx(&mut tx, message_id, actor).await?;
        let event = message_event(
            input.room_id,
            next_seq,
            ChatEventPayload::MessageCreated {
                message: message.clone(),
            },
        );
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_message",
            &message_id.to_string(),
            input.room_id,
            Some(next_seq),
            "chat.message.created",
            &event,
        )
        .await?;
        tx.commit().await?;
        Ok(SendMessageCommit {
            message,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn edit_message(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        input: EditMessage,
    ) -> ChatResult<MessageMutationCommit> {
        let mut tx = pool.begin().await?;
        let access = load_room_access(&mut tx, input.room_id, organization_id, actor).await?;
        let row = sqlx::query(
            "SELECT sender_id, room_seq FROM athena.chat_messages WHERE id = $1 AND room_id = $2 AND deleted_at IS NULL",
        )
            .bind(input.message_id)
            .bind(input.room_id)
            .fetch_optional(&mut *tx)
            .await?;
        let Some(row) = row else {
            return Err(ChatError::NotFound("message not found".to_string()));
        };
        let sender_id: String = row.try_get("sender_id")?;
        let room_seq: i64 = row.try_get("room_seq")?;
        ensure(
            sender_id == actor || actor_can_manage_members(&access.role),
            "message edit forbidden",
        )?;

        sqlx::query(
            r#"
            UPDATE athena.chat_messages
            SET body_text = $2,
                body_json = $3,
                metadata_json = $4,
                edited_at = now()
            WHERE id = $1
            "#,
        )
        .bind(input.message_id)
        .bind(&input.body_text)
        .bind(&input.body_json)
        .bind(&input.metadata_json)
        .execute(&mut *tx)
        .await?;

        let message = load_message_view_tx(&mut tx, input.message_id, actor).await?;
        let event = message_event(
            input.room_id,
            room_seq,
            ChatEventPayload::MessageUpdated {
                message: message.clone(),
            },
        );
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_message",
            &input.message_id.to_string(),
            input.room_id,
            Some(room_seq),
            "chat.message.updated",
            &event,
        )
        .await?;
        tx.commit().await?;
        Ok(MessageMutationCommit {
            message,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn delete_message(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        room_id: RoomId,
        message_id: MessageId,
    ) -> ChatResult<DeleteMessageCommit> {
        let mut tx = pool.begin().await?;
        let access = load_room_access(&mut tx, room_id, organization_id, actor).await?;
        let row = sqlx::query(
            "SELECT sender_id, room_seq FROM athena.chat_messages WHERE id = $1 AND room_id = $2",
        )
        .bind(message_id)
        .bind(room_id)
        .fetch_optional(&mut *tx)
        .await?;
        let Some(row) = row else {
            return Err(ChatError::NotFound("message not found".to_string()));
        };
        let sender_id: String = row.try_get("sender_id")?;
        let room_seq: i64 = row.try_get("room_seq")?;
        ensure(
            sender_id == actor || actor_can_manage_members(&access.role),
            "message delete forbidden",
        )?;

        sqlx::query(
            r#"
            UPDATE athena.chat_messages
            SET deleted_at = now(),
                body_text = NULL,
                body_json = NULL
            WHERE id = $1
            "#,
        )
        .bind(message_id)
        .execute(&mut *tx)
        .await?;

        let event = message_event(
            room_id,
            room_seq,
            ChatEventPayload::MessageDeleted { message_id },
        );
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_message",
            &message_id.to_string(),
            room_id,
            Some(room_seq),
            "chat.message.deleted",
            &event,
        )
        .await?;
        tx.commit().await?;
        Ok(DeleteMessageCommit {
            room_id,
            room_seq,
            message_id,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn mark_read_up_to(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        room_id: RoomId,
        seq: i64,
        message_id: Option<MessageId>,
    ) -> ChatResult<ReadCursorCommit> {
        let mut tx = pool.begin().await?;
        let access = load_room_access(&mut tx, room_id, organization_id, actor).await?;
        let message_cursor = if let Some(target_message_id) = message_id {
            sqlx::query(
                "SELECT id, room_seq FROM athena.chat_messages WHERE id = $1 AND room_id = $2",
            )
            .bind(target_message_id)
            .bind(room_id)
            .fetch_optional(&mut *tx)
            .await?
            .map(|row| -> ChatResult<(MessageId, i64)> {
                Ok((row.try_get("id")?, row.try_get("room_seq")?))
            })
            .transpose()?
        } else {
            None
        };
        let requested_seq = if seq > 0 {
            seq
        } else if let Some((_, room_seq)) = message_cursor {
            room_seq
        } else {
            0
        };
        let resolved_seq = requested_seq.min(access.last_message_seq).max(0);
        let resolved_message_id =
            message_cursor.and_then(|(target_message_id, target_room_seq)| {
                (target_room_seq == resolved_seq).then_some(target_message_id)
            });
        let cursor = sqlx::query(
            r#"
            UPDATE athena.chat_room_members
            SET last_read_seq = GREATEST(last_read_seq, $3),
                last_read_message_id = CASE
                    WHEN $4 IS NOT NULL AND $3 > last_read_seq THEN $4
                    ELSE last_read_message_id
                END
            WHERE room_id = $1
              AND user_id = $2
            RETURNING last_read_seq, last_read_message_id
            "#,
        )
        .bind(room_id)
        .bind(actor)
        .bind(resolved_seq)
        .bind(resolved_message_id)
        .fetch_one(&mut *tx)
        .await?;
        let committed_read_seq: i64 = cursor.try_get("last_read_seq")?;
        let committed_message_id: Option<MessageId> = cursor.try_get("last_read_message_id")?;

        let event = message_event(
            room_id,
            access.last_message_seq,
            ChatEventPayload::ReadUpdated {
                user_id: actor.to_string(),
                read_seq: committed_read_seq,
            },
        );
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "read_cursor",
            actor,
            room_id,
            Some(access.last_message_seq),
            "chat.read.updated",
            &event,
        )
        .await?;
        tx.commit().await?;
        Ok(ReadCursorCommit {
            room_id,
            user_id: actor.to_string(),
            last_read_seq: committed_read_seq,
            last_read_message_id: committed_message_id,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn add_reaction(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        message_id: MessageId,
        emoji: &str,
    ) -> ChatResult<ReactionCommit> {
        let mut tx = pool.begin().await?;
        let room = sqlx::query(
            r#"
            SELECT m.room_id, m.room_seq
            FROM athena.chat_messages AS m
            INNER JOIN athena.chat_rooms AS r
                ON r.id = m.room_id
            INNER JOIN athena.chat_room_members AS rm
                ON rm.room_id = r.id
            WHERE m.id = $1
              AND r.organization_id = $2
              AND rm.user_id = $3
            "#,
        )
        .bind(message_id)
        .bind(organization_id)
        .bind(actor)
        .fetch_optional(&mut *tx)
        .await?;
        let Some(room) = room else {
            return Err(ChatError::NotFound(
                "message not found or inaccessible".to_string(),
            ));
        };
        let room_id: RoomId = room.try_get("room_id")?;
        let room_seq: i64 = room.try_get("room_seq")?;
        sqlx::query(
            r#"
            INSERT INTO athena.chat_message_reactions
                (message_id, user_id, emoji)
            VALUES ($1, $2, $3)
            ON CONFLICT (message_id, user_id, emoji) DO NOTHING
            "#,
        )
        .bind(message_id)
        .bind(actor)
        .bind(emoji)
        .execute(&mut *tx)
        .await?;
        let summary = load_reaction_summary_tx(&mut tx, message_id, actor).await?;
        let event = message_event(
            room_id,
            room_seq,
            ChatEventPayload::ReactionUpdated {
                summary: summary.clone(),
            },
        );
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_message",
            &message_id.to_string(),
            room_id,
            Some(room_seq),
            "chat.reaction.updated",
            &event,
        )
        .await?;
        tx.commit().await?;
        Ok(ReactionCommit {
            room_id,
            summary,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn remove_reaction(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        message_id: MessageId,
        emoji: &str,
    ) -> ChatResult<ReactionCommit> {
        let mut tx = pool.begin().await?;
        let room = sqlx::query(
            r#"
            SELECT m.room_id, m.room_seq
            FROM athena.chat_messages AS m
            INNER JOIN athena.chat_rooms AS r
                ON r.id = m.room_id
            INNER JOIN athena.chat_room_members AS rm
                ON rm.room_id = r.id
            WHERE m.id = $1
              AND r.organization_id = $2
              AND rm.user_id = $3
            "#,
        )
        .bind(message_id)
        .bind(organization_id)
        .bind(actor)
        .fetch_optional(&mut *tx)
        .await?;
        let Some(room) = room else {
            return Err(ChatError::NotFound(
                "message not found or inaccessible".to_string(),
            ));
        };
        let room_id: RoomId = room.try_get("room_id")?;
        let room_seq: i64 = room.try_get("room_seq")?;
        sqlx::query("DELETE FROM athena.chat_message_reactions WHERE message_id = $1 AND user_id = $2 AND emoji = $3")
            .bind(message_id)
            .bind(actor)
            .bind(emoji)
            .execute(&mut *tx)
            .await?;
        let summary = load_reaction_summary_tx(&mut tx, message_id, actor).await?;
        let event = message_event(
            room_id,
            room_seq,
            ChatEventPayload::ReactionUpdated {
                summary: summary.clone(),
            },
        );
        let outbox_id = insert_chat_outbox_tx(
            &mut tx,
            "chat_message",
            &message_id.to_string(),
            room_id,
            Some(room_seq),
            "chat.reaction.updated",
            &event,
        )
        .await?;
        tx.commit().await?;
        Ok(ReactionCommit {
            room_id,
            summary,
            event,
            outbox_id: Some(outbox_id),
        })
    }

    async fn search_messages(
        &self,
        pool: &PgPool,
        organization_id: &str,
        actor: &str,
        input: SearchMessages,
    ) -> ChatResult<SearchPage> {
        let limit = input.limit.unwrap_or(25).clamp(1, 100);
        let rows = sqlx::query(
            r#"
            SELECT m.id, m.room_id
            FROM athena.chat_messages AS m
            INNER JOIN athena.chat_rooms AS r
                ON r.id = m.room_id
            INNER JOIN athena.chat_room_members AS rm
                ON rm.room_id = r.id
            WHERE r.organization_id = $1
              AND rm.user_id = $2
              AND ($3::uuid IS NULL OR m.room_id = $3)
              AND COALESCE(m.body_text, '') ILIKE '%' || $4 || '%'
            ORDER BY m.created_at DESC
            LIMIT $5
            "#,
        )
        .bind(organization_id)
        .bind(actor)
        .bind(input.room_id)
        .bind(input.query)
        .bind(limit)
        .fetch_all(pool)
        .await?;

        let mut items = Vec::with_capacity(rows.len());
        for row in rows {
            let id: MessageId = row.try_get("id")?;
            let room_id: RoomId = row.try_get("room_id")?;
            items.push(SearchHit {
                room_id,
                message: load_message_view(pool, id, actor).await?,
            });
        }
        Ok(SearchPage { items })
    }

    async fn lease_pending_outbox(
        &self,
        pool: &PgPool,
        limit: i64,
    ) -> ChatResult<Vec<(Uuid, RoomId, ChatEventEnvelope)>> {
        let rows = sqlx::query(
            r#"
            UPDATE athena.chat_outbox
            SET publish_attempts = publish_attempts + 1
            WHERE id IN (
                SELECT id
                FROM athena.chat_outbox
                WHERE published_at IS NULL
                ORDER BY created_at ASC
                LIMIT $1
                FOR UPDATE SKIP LOCKED
            )
            RETURNING id, room_id, payload_json
            "#,
        )
        .bind(limit)
        .fetch_all(pool)
        .await?;

        let mut events = Vec::with_capacity(rows.len());
        for row in rows {
            let payload: sqlx::types::Json<ChatEventEnvelope> = row.try_get("payload_json")?;
            events.push((row.try_get("id")?, row.try_get("room_id")?, payload.0));
        }
        Ok(events)
    }

    async fn mark_outbox_published(&self, pool: &PgPool, id: Uuid) -> ChatResult<()> {
        sqlx::query("UPDATE athena.chat_outbox SET published_at = now() WHERE id = $1")
            .bind(id)
            .execute(pool)
            .await?;
        Ok(())
    }

    async fn mark_outbox_failed(&self, pool: &PgPool, id: Uuid) -> ChatResult<()> {
        sqlx::query("UPDATE athena.chat_outbox SET published_at = NULL WHERE id = $1")
            .bind(id)
            .execute(pool)
            .await?;
        Ok(())
    }
}
