Files
fitaiProto/apps/admin/src/lib/chat.ts
T
2026-04-03 03:25:11 +02:00

825 lines
21 KiB
TypeScript

import { db, sql } from "@fitai/database";
import type { UserRole } from "@fitai/shared";
import log from "@/lib/logger";
export interface ChatUserContext {
id: string;
role: UserRole;
gymId?: string | null;
}
export interface ChatThreadSummary {
id: string;
type: "gym" | "dm";
gymId: string | null;
trainerId: string | null;
clientId: string | null;
createdBy: string;
lastMessageAt: Date | null;
createdAt: Date;
updatedAt: Date;
archivedAt: Date | null;
unreadCount: number;
lastMessageBody: string | null;
}
export interface ChatMessageRecord {
id: string;
threadId: string;
senderUserId: string;
body: string;
kind: "text" | "system";
attachments: Array<{ url: string; type: string; name?: string }>;
clientMessageId: string | null;
createdAt: Date;
editedAt: Date | null;
deletedAt: Date | null;
}
class ChatError extends Error {
status: number;
constructor(message: string, status: number) {
super(message);
this.status = status;
}
}
let tablesEnsured = false;
export async function ensureChatTables(): Promise<void> {
if (tablesEnsured) {
return;
}
await db.run(sql`
CREATE TABLE IF NOT EXISTS chat_threads (
id TEXT PRIMARY KEY,
type TEXT NOT NULL CHECK (type IN ('gym', 'dm')),
gym_id TEXT,
trainer_id TEXT,
client_id TEXT,
created_by TEXT NOT NULL,
last_message_at INTEGER,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
archived_at INTEGER,
FOREIGN KEY (gym_id) REFERENCES gyms(id) ON DELETE CASCADE,
FOREIGN KEY (trainer_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY (client_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY (created_by) REFERENCES users(id) ON DELETE CASCADE
)
`);
await db.run(sql`
CREATE TABLE IF NOT EXISTS chat_thread_members (
id TEXT PRIMARY KEY,
thread_id TEXT NOT NULL,
user_id TEXT NOT NULL,
role_in_thread TEXT NOT NULL CHECK (role_in_thread IN ('superAdmin', 'admin', 'trainer', 'client')),
joined_at INTEGER NOT NULL,
left_at INTEGER,
muted INTEGER NOT NULL DEFAULT 0,
last_read_message_id TEXT,
last_read_at INTEGER,
FOREIGN KEY (thread_id) REFERENCES chat_threads(id) ON DELETE CASCADE,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
UNIQUE(thread_id, user_id)
)
`);
await db.run(sql`
CREATE TABLE IF NOT EXISTS chat_messages (
id TEXT PRIMARY KEY,
thread_id TEXT NOT NULL,
sender_user_id TEXT NOT NULL,
body TEXT NOT NULL,
kind TEXT NOT NULL CHECK (kind IN ('text', 'system')) DEFAULT 'text',
attachments_json TEXT,
client_message_id TEXT,
created_at INTEGER NOT NULL,
edited_at INTEGER,
deleted_at INTEGER,
FOREIGN KEY (thread_id) REFERENCES chat_threads(id) ON DELETE CASCADE,
FOREIGN KEY (sender_user_id) REFERENCES users(id) ON DELETE CASCADE,
UNIQUE(sender_user_id, client_message_id)
)
`);
await db.run(sql`
CREATE TABLE IF NOT EXISTS chat_message_reads (
id TEXT PRIMARY KEY,
message_id TEXT NOT NULL,
user_id TEXT NOT NULL,
read_at INTEGER NOT NULL,
FOREIGN KEY (message_id) REFERENCES chat_messages(id) ON DELETE CASCADE,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
UNIQUE(message_id, user_id)
)
`);
await db.run(
sql`CREATE INDEX IF NOT EXISTS chat_threads_gym_type_last_msg_idx ON chat_threads(gym_id, type, last_message_at)`,
);
await db.run(
sql`CREATE INDEX IF NOT EXISTS chat_threads_type_idx ON chat_threads(type)`,
);
await db.run(
sql`CREATE UNIQUE INDEX IF NOT EXISTS chat_threads_trainer_client_type_idx ON chat_threads(trainer_id, client_id, type)`,
);
await db.run(
sql`CREATE INDEX IF NOT EXISTS chat_thread_members_user_thread_idx ON chat_thread_members(user_id, thread_id)`,
);
await db.run(
sql`CREATE INDEX IF NOT EXISTS chat_messages_thread_created_at_idx ON chat_messages(thread_id, created_at)`,
);
await db.run(
sql`CREATE INDEX IF NOT EXISTS chat_message_reads_user_read_at_idx ON chat_message_reads(user_id, read_at)`,
);
tablesEnsured = true;
}
export async function getOrCreateGymRoomThread(
currentUser: ChatUserContext,
): Promise<ChatThreadSummary> {
await ensureChatTables();
if (!currentUser.gymId) {
throw new ChatError("No gym assigned", 400);
}
const rows = await db.all(sql`
SELECT
id,
type,
gym_id as gymId,
trainer_id as trainerId,
client_id as clientId
FROM chat_threads
WHERE type = 'gym'
AND gym_id = ${currentUser.gymId}
AND archived_at IS NULL
LIMIT 1
`);
let thread = rows?.[0] as ThreadRow | undefined;
if (!thread) {
const now = nowEpochSeconds();
const newThreadId = generateId("thread");
await db.run(sql`
INSERT INTO chat_threads (
id,
type,
gym_id,
created_by,
created_at,
updated_at
) VALUES (
${newThreadId},
${"gym"},
${currentUser.gymId},
${currentUser.id},
${now},
${now}
)
`);
const createdRows = await db.all(sql`
SELECT
id,
type,
gym_id as gymId,
trainer_id as trainerId,
client_id as clientId
FROM chat_threads
WHERE id = ${newThreadId}
LIMIT 1
`);
thread = createdRows?.[0] as ThreadRow | undefined;
}
if (!thread) {
throw new ChatError("Failed to initialize gym room", 500);
}
await ensureThreadMember(thread.id, currentUser.id, currentUser.role);
return await getThreadSummaryForUser(currentUser.id, thread.id);
}
export async function getOrCreateDmThread(params: {
trainerId: string;
clientId: string;
createdBy: ChatUserContext;
}): Promise<ChatThreadSummary> {
await ensureChatTables();
const trainerRows = await db.all(sql`
SELECT id, role, gym_id as gymId
FROM users
WHERE id = ${params.trainerId}
LIMIT 1
`);
const clientRows = await db.all(sql`
SELECT id, role, gym_id as gymId
FROM users
WHERE id = ${params.clientId}
LIMIT 1
`);
const trainer = trainerRows?.[0] as
| { id: string; role: UserRole; gymId: string | null }
| undefined;
const client = clientRows?.[0] as
| { id: string; role: UserRole; gymId: string | null }
| undefined;
if (!trainer || !client) {
throw new ChatError("Trainer or client not found", 404);
}
if (client.role !== "client") {
throw new ChatError("Direct message target must be a client", 400);
}
if (trainer.role !== "trainer" && trainer.role !== "admin") {
throw new ChatError("Direct message trainer is invalid", 400);
}
if (!trainer.gymId || trainer.gymId !== client.gymId) {
throw new ChatError("Trainer and client must belong to same gym", 400);
}
const assignmentRows = await db.all(sql`
SELECT id
FROM trainer_client_assignments
WHERE trainer_id = ${params.trainerId}
AND client_id = ${params.clientId}
AND is_active = 1
LIMIT 1
`);
if (!assignmentRows?.[0]) {
throw new ChatError("Trainer is not assigned to this client", 403);
}
const rows = await db.all(sql`
SELECT
id,
type,
gym_id as gymId,
trainer_id as trainerId,
client_id as clientId
FROM chat_threads
WHERE type = 'dm'
AND trainer_id = ${params.trainerId}
AND client_id = ${params.clientId}
AND archived_at IS NULL
LIMIT 1
`);
let thread = rows?.[0] as ThreadRow | undefined;
if (!thread) {
const now = nowEpochSeconds();
const threadId = generateId("thread");
await db.run(sql`
INSERT INTO chat_threads (
id,
type,
gym_id,
trainer_id,
client_id,
created_by,
created_at,
updated_at
) VALUES (
${threadId},
${"dm"},
${trainer.gymId},
${params.trainerId},
${params.clientId},
${params.createdBy.id},
${now},
${now}
)
`);
await ensureThreadMember(threadId, params.trainerId, trainer.role);
await ensureThreadMember(threadId, params.clientId, client.role);
const createdRows = await db.all(sql`
SELECT
id,
type,
gym_id as gymId,
trainer_id as trainerId,
client_id as clientId
FROM chat_threads
WHERE id = ${threadId}
LIMIT 1
`);
thread = createdRows?.[0] as ThreadRow | undefined;
}
if (!thread) {
throw new ChatError("Failed to initialize direct message", 500);
}
await ensureThreadMember(
thread.id,
params.createdBy.id,
params.createdBy.role,
);
return await getThreadSummaryForUser(params.createdBy.id, thread.id);
}
export async function listThreadsForUser(
currentUser: ChatUserContext,
): Promise<ChatThreadSummary[]> {
await ensureChatTables();
const rows = await db.all(sql`
SELECT
t.id,
t.type,
t.gym_id as gymId,
t.trainer_id as trainerId,
t.client_id as clientId,
t.created_by as createdBy,
t.last_message_at as lastMessageAt,
t.created_at as createdAt,
t.updated_at as updatedAt,
t.archived_at as archivedAt,
(
SELECT m.body
FROM chat_messages m
WHERE m.thread_id = t.id
AND m.deleted_at IS NULL
ORDER BY m.created_at DESC
LIMIT 1
) as lastMessageBody,
(
SELECT COUNT(*)
FROM chat_messages m
WHERE m.thread_id = t.id
AND m.deleted_at IS NULL
AND m.sender_user_id != ${currentUser.id}
AND m.created_at > COALESCE(mem.last_read_at, 0)
) as unreadCount
FROM chat_threads t
JOIN chat_thread_members mem
ON mem.thread_id = t.id
WHERE mem.user_id = ${currentUser.id}
AND mem.left_at IS NULL
AND t.archived_at IS NULL
ORDER BY COALESCE(t.last_message_at, t.created_at) DESC
`);
return (rows as ThreadSummaryRow[]).map(mapThreadSummaryRow);
}
export async function getThreadMessages(params: {
currentUser: ChatUserContext;
threadId: string;
cursor?: number;
limit?: number;
}): Promise<{ messages: ChatMessageRecord[]; nextCursor: number | null }> {
await ensureChatTables();
await assertThreadAccess(params.currentUser, params.threadId);
const pageSize = Math.max(1, Math.min(100, params.limit ?? 30));
const cursor = params.cursor;
const rows = await db.all(sql`
SELECT
id,
thread_id as threadId,
sender_user_id as senderUserId,
body,
kind,
attachments_json as attachmentsJson,
client_message_id as clientMessageId,
created_at as createdAt,
edited_at as editedAt,
deleted_at as deletedAt
FROM chat_messages
WHERE thread_id = ${params.threadId}
AND deleted_at IS NULL
${cursor ? sql`AND created_at < ${cursor}` : sql``}
ORDER BY created_at DESC
LIMIT ${pageSize + 1}
`);
const data = rows as MessageRow[];
const hasNextPage = data.length > pageSize;
const pageRows = hasNextPage ? data.slice(0, pageSize) : data;
const messages = pageRows.map(mapMessageRow).reverse();
const nextCursor =
hasNextPage && pageRows.length > 0
? Number(pageRows[pageRows.length - 1].createdAt)
: null;
return { messages, nextCursor };
}
export async function createThreadMessage(params: {
currentUser: ChatUserContext;
threadId: string;
body: string;
clientMessageId?: string;
attachments?: Array<{ url: string; type: string; name?: string }>;
}): Promise<ChatMessageRecord> {
await ensureChatTables();
await assertThreadAccess(params.currentUser, params.threadId);
const body = params.body.trim();
if (!body) {
throw new ChatError("Message body is required", 400);
}
if (body.length > 2000) {
throw new ChatError("Message is too long", 400);
}
if (params.clientMessageId) {
const existingRows = await db.all(sql`
SELECT
id,
thread_id as threadId,
sender_user_id as senderUserId,
body,
kind,
attachments_json as attachmentsJson,
client_message_id as clientMessageId,
created_at as createdAt,
edited_at as editedAt,
deleted_at as deletedAt
FROM chat_messages
WHERE sender_user_id = ${params.currentUser.id}
AND client_message_id = ${params.clientMessageId}
LIMIT 1
`);
if (existingRows?.[0]) {
return mapMessageRow(existingRows[0] as MessageRow);
}
}
const messageId = generateId("msg");
const now = nowEpochSeconds();
await db.run(sql`
INSERT INTO chat_messages (
id,
thread_id,
sender_user_id,
body,
kind,
attachments_json,
client_message_id,
created_at
) VALUES (
${messageId},
${params.threadId},
${params.currentUser.id},
${body},
${"text"},
${params.attachments?.length ? JSON.stringify(params.attachments) : null},
${params.clientMessageId ?? null},
${now}
)
`);
await db.run(sql`
UPDATE chat_threads
SET
last_message_at = ${now},
updated_at = ${now}
WHERE id = ${params.threadId}
`);
const messageRows = await db.all(sql`
SELECT
id,
thread_id as threadId,
sender_user_id as senderUserId,
body,
kind,
attachments_json as attachmentsJson,
client_message_id as clientMessageId,
created_at as createdAt,
edited_at as editedAt,
deleted_at as deletedAt
FROM chat_messages
WHERE id = ${messageId}
LIMIT 1
`);
const message = messageRows?.[0] as MessageRow | undefined;
if (!message) {
throw new ChatError("Failed to send message", 500);
}
return mapMessageRow(message);
}
export async function markThreadAsRead(params: {
currentUser: ChatUserContext;
threadId: string;
lastReadMessageId?: string;
}): Promise<void> {
await ensureChatTables();
await assertThreadAccess(params.currentUser, params.threadId);
let targetMessageId = params.lastReadMessageId ?? null;
if (!targetMessageId) {
const latestRows = await db.all(sql`
SELECT id
FROM chat_messages
WHERE thread_id = ${params.threadId}
AND deleted_at IS NULL
ORDER BY created_at DESC
LIMIT 1
`);
const latest = latestRows?.[0] as { id?: string } | undefined;
targetMessageId = latest?.id ?? null;
}
const now = nowEpochSeconds();
await db.run(sql`
UPDATE chat_thread_members
SET
last_read_message_id = ${targetMessageId},
last_read_at = ${now}
WHERE thread_id = ${params.threadId}
AND user_id = ${params.currentUser.id}
`);
}
export function isChatError(error: unknown): error is ChatError {
return error instanceof ChatError;
}
async function assertThreadAccess(
currentUser: ChatUserContext,
threadId: string,
): Promise<ThreadRow> {
const rows = await db.all(sql`
SELECT
id,
type,
gym_id as gymId,
trainer_id as trainerId,
client_id as clientId
FROM chat_threads
WHERE id = ${threadId}
AND archived_at IS NULL
LIMIT 1
`);
const thread = rows?.[0] as ThreadRow | undefined;
if (!thread) {
throw new ChatError("Thread not found", 404);
}
if (currentUser.role === "superAdmin") {
return thread;
}
if (thread.type === "gym") {
if (!currentUser.gymId || currentUser.gymId !== thread.gymId) {
throw new ChatError("Forbidden", 403);
}
await ensureThreadMember(thread.id, currentUser.id, currentUser.role);
return thread;
}
const isParticipant =
thread.trainerId === currentUser.id || thread.clientId === currentUser.id;
if (!isParticipant) {
throw new ChatError("Forbidden", 403);
}
if (thread.trainerId && thread.clientId) {
const assignmentRows = await db.all(sql`
SELECT id
FROM trainer_client_assignments
WHERE trainer_id = ${thread.trainerId}
AND client_id = ${thread.clientId}
AND is_active = 1
LIMIT 1
`);
if (!assignmentRows?.[0]) {
throw new ChatError("Trainer-client assignment is not active", 403);
}
}
await ensureThreadMember(thread.id, currentUser.id, currentUser.role);
return thread;
}
async function getThreadSummaryForUser(
userId: string,
threadId: string,
): Promise<ChatThreadSummary> {
const rows = await db.all(sql`
SELECT
t.id,
t.type,
t.gym_id as gymId,
t.trainer_id as trainerId,
t.client_id as clientId,
t.created_by as createdBy,
t.last_message_at as lastMessageAt,
t.created_at as createdAt,
t.updated_at as updatedAt,
t.archived_at as archivedAt,
(
SELECT m.body
FROM chat_messages m
WHERE m.thread_id = t.id
AND m.deleted_at IS NULL
ORDER BY m.created_at DESC
LIMIT 1
) as lastMessageBody,
(
SELECT COUNT(*)
FROM chat_messages m
WHERE m.thread_id = t.id
AND m.deleted_at IS NULL
AND m.sender_user_id != ${userId}
AND m.created_at > COALESCE(mem.last_read_at, 0)
) as unreadCount
FROM chat_threads t
JOIN chat_thread_members mem
ON mem.thread_id = t.id
AND mem.user_id = ${userId}
AND mem.left_at IS NULL
WHERE t.id = ${threadId}
LIMIT 1
`);
const row = rows?.[0] as ThreadSummaryRow | undefined;
if (!row) {
throw new ChatError("Thread not found", 404);
}
return mapThreadSummaryRow(row);
}
async function ensureThreadMember(
threadId: string,
userId: string,
role: UserRole,
): Promise<void> {
const now = nowEpochSeconds();
await db.run(sql`
INSERT OR IGNORE INTO chat_thread_members (
id,
thread_id,
user_id,
role_in_thread,
joined_at,
muted
) VALUES (
${generateId("member")},
${threadId},
${userId},
${role},
${now},
${0}
)
`);
await db.run(sql`
UPDATE chat_thread_members
SET left_at = NULL
WHERE thread_id = ${threadId}
AND user_id = ${userId}
`);
}
function mapThreadSummaryRow(row: ThreadSummaryRow): ChatThreadSummary {
return {
id: row.id,
type: row.type,
gymId: row.gymId ?? null,
trainerId: row.trainerId ?? null,
clientId: row.clientId ?? null,
createdBy: row.createdBy,
lastMessageAt: toDate(row.lastMessageAt),
createdAt: toDate(row.createdAt) ?? new Date(),
updatedAt: toDate(row.updatedAt) ?? new Date(),
archivedAt: toDate(row.archivedAt),
unreadCount: Number(row.unreadCount ?? 0),
lastMessageBody: row.lastMessageBody ?? null,
};
}
function mapMessageRow(row: MessageRow): ChatMessageRecord {
return {
id: row.id,
threadId: row.threadId,
senderUserId: row.senderUserId,
body: row.body,
kind: row.kind,
attachments: parseAttachments(row.attachmentsJson),
clientMessageId: row.clientMessageId ?? null,
createdAt: toDate(row.createdAt) ?? new Date(),
editedAt: toDate(row.editedAt),
deletedAt: toDate(row.deletedAt),
};
}
function parseAttachments(
value: string | null,
): Array<{ url: string; type: string; name?: string }> {
if (!value) {
return [];
}
try {
const parsed = JSON.parse(value);
if (!Array.isArray(parsed)) {
return [];
}
return parsed.filter(
(item): item is { url: string; type: string; name?: string } =>
item &&
typeof item === "object" &&
typeof item.url === "string" &&
typeof item.type === "string",
);
} catch (error) {
log.warn("Failed to parse chat attachments JSON", { value, error });
return [];
}
}
function toDate(value: number | string | null | undefined): Date | null {
if (value === null || value === undefined) {
return null;
}
const numberValue = Number(value);
if (!Number.isFinite(numberValue)) {
return null;
}
return new Date(numberValue * 1000);
}
function generateId(prefix: string): string {
return `${prefix}_${Date.now()}_${Math.random().toString(36).slice(2, 10)}`;
}
function nowEpochSeconds(): number {
return Math.floor(Date.now() / 1000);
}
interface ThreadRow {
id: string;
type: "gym" | "dm";
gymId: string | null;
trainerId: string | null;
clientId: string | null;
}
interface ThreadSummaryRow {
id: string;
type: "gym" | "dm";
gymId: string | null;
trainerId: string | null;
clientId: string | null;
createdBy: string;
lastMessageAt: number | null;
createdAt: number;
updatedAt: number;
archivedAt: number | null;
unreadCount: number;
lastMessageBody: string | null;
}
interface MessageRow {
id: string;
threadId: string;
senderUserId: string;
body: string;
kind: "text" | "system";
attachmentsJson: string | null;
clientMessageId: string | null;
createdAt: number;
editedAt: number | null;
deletedAt: number | null;
}