psg-backchannel/backchannel-server/src/db/messages.rs

use backchannel_common::protocol::server::{HistoricChannelMessage, HistoricDm};
use chrono::{DateTime, Utc};
use sqlx::SqlitePool;
use uuid::Uuid;

use crate::error::{Result, ServerError};

#[derive(Debug, sqlx::FromRow)]
struct ChannelMessageRow {
    pub id: String,
    pub channel_id: String,
    pub author_id: String,
    pub author_username: String,
    pub content: String,
    pub created_at: String,
}

#[derive(Debug, sqlx::FromRow)]
struct DmRow {
    pub id: String,
    pub sender_id: String,
    pub ciphertext: String,
    pub nonce: String,
    pub created_at: String,
}

pub async fn insert_channel_message(
    pool: &SqlitePool,
    channel_id: Uuid,
    author_id: Uuid,
    content: &str,
) -> Result<(Uuid, DateTime<Utc>)> {
    let id = Uuid::new_v4();
    let now = Utc::now();
    let now_str = now.to_rfc3339();

    sqlx::query(
        "INSERT INTO channel_messages (id, channel_id, author_id, content, created_at)
         VALUES (?, ?, ?, ?, ?)",
    )
    .bind(id.to_string())
    .bind(channel_id.to_string())
    .bind(author_id.to_string())
    .bind(content)
    .bind(&now_str)
    .execute(pool)
    .await?;

    Ok((id, now))
}

pub async fn fetch_channel_history(
    pool: &SqlitePool,
    channel_id: Uuid,
    before_message_id: Option<Uuid>,
    limit: u32,
) -> Result<Vec<HistoricChannelMessage>> {
    let limit = limit.min(100) as i64;

    let rows: Vec<ChannelMessageRow> = if let Some(before_id) = before_message_id {
        // Fetch messages older than `before_id` by joining to get its created_at
        sqlx::query_as::<_, ChannelMessageRow>(
            "SELECT m.id, m.channel_id, m.author_id, u.username AS author_username,
                    m.content, m.created_at
             FROM channel_messages m
             JOIN users u ON u.id = m.author_id
             WHERE m.channel_id = ?
               AND m.created_at < (SELECT created_at FROM channel_messages WHERE id = ?)
             ORDER BY m.created_at DESC
             LIMIT ?",
        )
        .bind(channel_id.to_string())
        .bind(before_id.to_string())
        .bind(limit)
        .fetch_all(pool)
        .await?
    } else {
        sqlx::query_as::<_, ChannelMessageRow>(
            "SELECT m.id, m.channel_id, m.author_id, u.username AS author_username,
                    m.content, m.created_at
             FROM channel_messages m
             JOIN users u ON u.id = m.author_id
             WHERE m.channel_id = ?
             ORDER BY m.created_at DESC
             LIMIT ?",
        )
        .bind(channel_id.to_string())
        .bind(limit)
        .fetch_all(pool)
        .await?
    };

    let mut messages: Vec<HistoricChannelMessage> = rows
        .into_iter()
        .map(|r| {
            let ts = r
                .created_at
                .parse::<DateTime<Utc>>()
                .unwrap_or_else(|_| Utc::now());
            Ok(HistoricChannelMessage {
                message_id: Uuid::parse_str(&r.id)
                    .map_err(|e| ServerError::Internal(e.to_string()))?,
                author_id: Uuid::parse_str(&r.author_id)
                    .map_err(|e| ServerError::Internal(e.to_string()))?,
                author_username: r.author_username,
                content: r.content,
                timestamp: ts,
            })
        })
        .collect::<Result<Vec<_>>>()?;

    // Return in chronological order (oldest first).
    messages.reverse();
    Ok(messages)
}

pub async fn insert_dm(
    pool: &SqlitePool,
    sender_id: Uuid,
    recipient_id: Uuid,
    ciphertext: &str,
    nonce: &str,
) -> Result<(Uuid, DateTime<Utc>)> {
    let id = Uuid::new_v4();
    let now = Utc::now();
    let now_str = now.to_rfc3339();

    sqlx::query(
        "INSERT INTO direct_messages (id, sender_id, recipient_id, ciphertext, nonce, created_at)
         VALUES (?, ?, ?, ?, ?, ?)",
    )
    .bind(id.to_string())
    .bind(sender_id.to_string())
    .bind(recipient_id.to_string())
    .bind(ciphertext)
    .bind(nonce)
    .bind(&now_str)
    .execute(pool)
    .await?;

    Ok((id, now))
}

pub async fn fetch_dm_history(
    pool: &SqlitePool,
    user_a: Uuid,
    user_b: Uuid,
    before_message_id: Option<Uuid>,
    limit: u32,
) -> Result<Vec<HistoricDm>> {
    let limit = limit.min(100) as i64;

    let rows: Vec<DmRow> = if let Some(before_id) = before_message_id {
        sqlx::query_as::<_, DmRow>(
            "SELECT id, sender_id, ciphertext, nonce, created_at
             FROM direct_messages
             WHERE ((sender_id = ? AND recipient_id = ?)
                 OR (sender_id = ? AND recipient_id = ?))
               AND created_at < (SELECT created_at FROM direct_messages WHERE id = ?)
             ORDER BY created_at DESC
             LIMIT ?",
        )
        .bind(user_a.to_string())
        .bind(user_b.to_string())
        .bind(user_b.to_string())
        .bind(user_a.to_string())
        .bind(before_id.to_string())
        .bind(limit)
        .fetch_all(pool)
        .await?
    } else {
        sqlx::query_as::<_, DmRow>(
            "SELECT id, sender_id, ciphertext, nonce, created_at
             FROM direct_messages
             WHERE (sender_id = ? AND recipient_id = ?)
                OR (sender_id = ? AND recipient_id = ?)
             ORDER BY created_at DESC
             LIMIT ?",
        )
        .bind(user_a.to_string())
        .bind(user_b.to_string())
        .bind(user_b.to_string())
        .bind(user_a.to_string())
        .bind(limit)
        .fetch_all(pool)
        .await?
    };

    let mut messages: Vec<HistoricDm> = rows
        .into_iter()
        .map(|r| {
            let ts = r
                .created_at
                .parse::<DateTime<Utc>>()
                .unwrap_or_else(|_| Utc::now());
            Ok(HistoricDm {
                message_id: Uuid::parse_str(&r.id)
                    .map_err(|e| ServerError::Internal(e.to_string()))?,
                sender_id: Uuid::parse_str(&r.sender_id)
                    .map_err(|e| ServerError::Internal(e.to_string()))?,
                ciphertext: r.ciphertext,
                nonce: r.nonce,
                timestamp: ts,
            })
        })
        .collect::<Result<Vec<_>>>()?;

    messages.reverse();
    Ok(messages)
}