MDietermann
099315a538
Messages were only reachable by polling. This adds the server half of the realtime path: a cursor that cannot skip rows, a change log, and a socket that pushes "conversation X is at seq N". The cursor is a counter on conversations, drawn with UPDATE ... RETURNING inside the mutating transaction. That takes the conversation row lock, so a second writer blocks until the first commits and commit order equals seq order. A BIGSERIAL on messages does not have this property: it hands out numbers at INSERT time while rows become visible at COMMIT time, and those two orders diverge under concurrent senders — a reader following seq > n then loses a message permanently. Two tests cover this; the second one asserts that the naive design *does* lose a row, so the first is proven to measure something. created_at loses its ordering role. It is NOW(), i.e. transaction start, so it is neither unique nor commit-ordered — the existing ORDER BY created_at was already wrong, independently of realtime. Every mutation appends to conversation_events, and that row is the event. A three-line trigger on that table fires pg_notify, so no future write path can forget to signal, and NOTIFY being transactional means it fires exactly when the row becomes visible. Fan-out always goes through NOTIFY, even with a single instance. One path instead of two means the tested path is the production path, and running more than one instance needs no second implementation. Per connection the pending work is a map rather than a queue: only the newest seq per conversation matters, so a newer value overwrites an older one, there is no overflow and no overflow error path. Entries are merged with max(), because NOTIFY only orders per listener. When the listener reconnects, every client is told to resync — notifications sent during the gap are gone, as NOTIFY keeps no history. Sockets are closed by hand on shutdown: axum's graceful shutdown neither waits for nor closes upgraded connections (axum#3003, fix PR #3831 still open). Also adds membership checks to the message history endpoint, which had none, and paginates it — it previously returned entire conversations unbounded. Not yet done, marked as TODO in the code: reactions and attachments do not emit events, and create/update/delete still trust user_id from the request body.
178 lines
5.2 KiB
Rust
178 lines
5.2 KiB
Rust
use serde::{Deserialize, Serialize};
|
|
use uuid::Uuid;
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct Claims {
|
|
pub sub: String,
|
|
pub exp: usize,
|
|
}
|
|
|
|
// ── Enums ────────────────────────────────────────────────────────────────────
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::Type)]
|
|
#[sqlx(type_name = "auth_token_kind", rename_all = "snake_case")]
|
|
pub enum AuthTokenKind {
|
|
PasswordReset,
|
|
EmailVerification,
|
|
RefreshToken,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::Type)]
|
|
#[sqlx(type_name = "workspace_role", rename_all = "snake_case")]
|
|
pub enum WorkspaceRole {
|
|
Owner,
|
|
Admin,
|
|
Member,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::Type)]
|
|
#[sqlx(type_name = "conversation_kind", rename_all = "snake_case")]
|
|
pub enum ConversationKind {
|
|
Dm,
|
|
Group,
|
|
Channel,
|
|
}
|
|
|
|
// ── Tables ───────────────────────────────────────────────────────────────────
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct User {
|
|
pub id: Uuid,
|
|
pub email: String,
|
|
pub password_hash: String,
|
|
pub username: String,
|
|
pub display_name: String,
|
|
pub avatar_url: Option<String>,
|
|
pub bio: Option<String>,
|
|
pub email_verified: bool,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
pub updated_at: chrono::DateTime<chrono::Utc>,
|
|
pub deleted_at: Option<chrono::DateTime<chrono::Utc>>,
|
|
}
|
|
|
|
#[derive(Debug, Serialize)]
|
|
pub struct UserResponse {
|
|
pub id: Uuid,
|
|
pub email: String,
|
|
pub username: String,
|
|
pub display_name: String,
|
|
pub avatar_url: Option<String>,
|
|
pub bio: Option<String>,
|
|
pub email_verified: bool,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
}
|
|
|
|
impl From<User> for UserResponse {
|
|
fn from(u: User) -> Self {
|
|
Self {
|
|
id: u.id,
|
|
email: u.email,
|
|
username: u.username,
|
|
display_name: u.display_name,
|
|
avatar_url: u.avatar_url,
|
|
bio: u.bio,
|
|
email_verified: u.email_verified,
|
|
created_at: u.created_at,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct AuthToken {
|
|
pub id: Uuid,
|
|
pub user_id: Uuid,
|
|
pub token_hash: String,
|
|
pub kind: AuthTokenKind,
|
|
pub expires_at: chrono::DateTime<chrono::Utc>,
|
|
pub used_at: Option<chrono::DateTime<chrono::Utc>>,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct Workspace {
|
|
pub id: Uuid,
|
|
pub name: String,
|
|
pub slug: String,
|
|
pub owner_id: Uuid,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
pub updated_at: chrono::DateTime<chrono::Utc>,
|
|
pub deleted_at: Option<chrono::DateTime<chrono::Utc>>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct WorkspaceMember {
|
|
pub workspace_id: Uuid,
|
|
pub user_id: Uuid,
|
|
pub role: WorkspaceRole,
|
|
pub display_name: Option<String>,
|
|
pub avatar_url: Option<String>,
|
|
pub joined_at: chrono::DateTime<chrono::Utc>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct WorkspaceInvitation {
|
|
pub id: Uuid,
|
|
pub workspace_id: Uuid,
|
|
pub created_by: Uuid,
|
|
pub token: String,
|
|
pub max_uses: Option<i32>,
|
|
pub use_count: i32,
|
|
pub expires_at: Option<chrono::DateTime<chrono::Utc>>,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct Conversation {
|
|
pub id: Uuid,
|
|
pub workspace_id: Option<Uuid>,
|
|
pub kind: ConversationKind,
|
|
pub name: Option<String>,
|
|
pub is_public: bool,
|
|
pub created_by: Uuid,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
pub updated_at: chrono::DateTime<chrono::Utc>,
|
|
pub deleted_at: Option<chrono::DateTime<chrono::Utc>>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct ConversationMember {
|
|
pub conversation_id: Uuid,
|
|
pub user_id: Uuid,
|
|
pub joined_at: chrono::DateTime<chrono::Utc>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct Message {
|
|
pub id: Uuid,
|
|
pub conversation_id: Uuid,
|
|
pub user_id: Uuid,
|
|
pub content: String,
|
|
/// Immutable sort key within the conversation. Ordering follows this, never
|
|
/// `created_at` — that is NOW(), i.e. transaction start, so it is neither
|
|
/// unique nor commit-ordered.
|
|
pub seq: i64,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
pub updated_at: chrono::DateTime<chrono::Utc>,
|
|
pub deleted_at: Option<chrono::DateTime<chrono::Utc>>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct Attachment {
|
|
pub id: Uuid,
|
|
pub message_id: Uuid,
|
|
pub url: String,
|
|
pub filename: String,
|
|
pub mime_type: String,
|
|
pub size_bytes: i64,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize, sqlx::FromRow)]
|
|
pub struct Reaction {
|
|
pub id: Uuid,
|
|
pub message_id: Uuid,
|
|
pub user_id: Uuid,
|
|
pub emoji: String,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
}
|