import { Pool, type QueryResultRow } from 'pg'; import { closeResolvedPool, resolvePool, type DatabaseConnection } from '../../db/pool.js'; export type ChatMediaType = 'image' | 'video' | null; export type ReportReason = 'harassment' | 'spam' | 'inappropriate_content' | 'other'; export interface ChatMessage { id: string; match_id: string; user_id: string; content: string | null; created_at: string; read_at: string | null; delivered_at: string | null; deleted_at: string | null; media_path: string | null; media_type: ChatMediaType; media_expires_at: string | null; media_expired: boolean; media_viewed_at: string | null; version: number; encrypted_content: string | null; encryption_iv: string | null; keys_metadata: Record | null; moderation_status: string | null; flag_reason: string | null; category: string | null; } interface MessageRecord extends QueryResultRow, Omit { created_at: Date; read_at: Date | null; delivered_at: Date | null; deleted_at: Date | null; media_expires_at: Date | null; media_viewed_at: Date | null; } export class ChatError extends Error { constructor( public readonly code: string, message: string, public readonly status: 400 | 403 | 404 | 409, ) { super(message); } } /** * The reply to a message poll. A request without a cursor is a full snapshot; a * request with one carries only what changed, so a quiet chat costs an empty * array rather than the whole backlog. */ export interface MessagesPage { /** Newest first. Messages created or changed since the cursor, or the snapshot. */ messages: ChatMessage[]; /** Messages the caller deleted for themselves since the cursor. */ removed_ids: string[]; /** Cursor to send as `since` on the next poll. */ server_time: string; /** True when `messages` is a complete snapshot the client may replace its cache with. */ complete: boolean; /** Present only when the caller asked for it, folding the typing poll into this one. */ typing?: PartnerTyping; } export interface PartnerTyping { typing: boolean; expires_at: string | null; } export interface ListMessagesOptions { /** ISO timestamp from a previous `server_time`. Omit for a full snapshot. */ since?: string; includeTyping?: boolean; } export interface ChatRepository { listMessages(userId: string, matchId: string, options?: ListMessagesOptions): Promise; sendText(userId: string, matchId: string, content: string): Promise; unreadCounts(userId: string): Promise<{ total: number; by_match: Record }>; markMatchRead(userId: string, matchId: string): Promise<{ updated: number; read_at: string }>; markDelivered(userId: string, messageId: string): Promise; deleteForSelf(userId: string, messageId: string): Promise; deleteForEveryone(userId: string, messageId: string): Promise; report(userId: string, messageId: string, reason: ReportReason): Promise; setTyping(userId: string, matchId: string, ttlMs: number): Promise; getPartnerTyping(userId: string, matchId: string): Promise; close(): Promise; } // `media_expired` is derived rather than read. Nothing ever writes the stored // column — it is inserted `false` and never updated — so a video that reached // `media_expires_at` still arrived flagged as playable, while the media itself // was already refused (see `accessible` in the media repository, which treats // `media_expired or media_expires_at <= now()` as gone). Deriving it here is the // same rule the media side already applies, and it is what makes the client's // "Video expired" state reachable at all. const columns = `m.id, m.match_id, m.user_id, m.content, m.created_at, m.read_at, m.delivered_at, m.deleted_at, m.media_path, m.media_type, m.media_expires_at, (m.media_expired or (m.media_expires_at is not null and m.media_expires_at <= now())) as media_expired, m.media_viewed_at, m.version, m.encrypted_content, m.encryption_iv, m.keys_metadata, m.moderation_status, m.flag_reason, m.category`; function iso(value: Date | null): string | null { return value?.toISOString() ?? null; } function toMessage(row: MessageRecord): ChatMessage { return { ...row, created_at: row.created_at.toISOString(), read_at: iso(row.read_at), delivered_at: iso(row.delivered_at), deleted_at: iso(row.deleted_at), media_expires_at: iso(row.media_expires_at), media_viewed_at: iso(row.media_viewed_at), }; } export class PostgresChatRepository implements ChatRepository { private readonly pool: Pool; private readonly ownsPool: boolean; constructor(connection: DatabaseConnection) { const resolved = resolvePool(connection); this.pool = resolved.pool; this.ownsPool = resolved.owned; } private async assertMatchAccess(userId: string, matchId: string): Promise { const result = await this.pool.query( `select 1 from profiles p join matches ma on ma.couple_id = p.couple_id where p.id = $1 and ma.id = $2`, [userId, matchId], ); if (!result.rowCount) throw new ChatError('match_not_found', 'Match not found', 404); } private async accessibleMessage(userId: string, messageId: string): Promise { const result = await this.pool.query( `select ${columns} from messages m join matches ma on ma.id = m.match_id join profiles p on p.couple_id = ma.couple_id where p.id = $1 and m.id = $2`, [userId, messageId], ); const row = result.rows[0]; if (!row) throw new ChatError('message_not_found', 'Message not found', 404); return row; } /** * Without `since` this is the previous behaviour: the newest 500 visible messages. * With `since` it returns only rows whose visible state moved after that instant — * new messages, existing ones whose delivery, read, deletion or media-view * timestamp advanced, and videos whose media expired inside the window — plus the * ids the caller has since deleted for themselves. * * Moderation metadata (`moderation_status`, `flag_reason`, `category`) is not * covered: the classifier and the admin tool rewrite those columns without * advancing any timestamp, so a delta cannot see them move. No client renders * them today, so nothing user-visible goes stale; giving them a cursor would * mean adding a column to `messages`. * Rows are returned whole rather than as patches so the client merges by id and * needs no second shape to interpret. * * The returned cursor is deliberately a couple of seconds behind the server clock: * a transaction that started before this read but committed after it would * otherwise fall into the gap between two polls. The overlap re-delivers a handful * of rows the client already has, which its merge-by-id absorbs. */ async listMessages(userId: string, matchId: string, options: ListMessagesOptions = {}): Promise { await this.assertMatchAccess(userId, matchId); const since = options.since ?? null; const visible = `not exists (select 1 from message_deletions d where d.message_id = m.id and d.user_id = $2)`; // Expiry is the one visible change no row write announces: the clock crosses // `media_expires_at` and nothing is updated, so no timestamp moves past the // cursor. Comparing the deadline against the window instead emits the row in // the single poll whose `(since, now()]` contains it, which is exactly when // the derived `media_expired` above flips. const changed = `(m.created_at > $3 or m.read_at > $3 or m.delivered_at > $3 or m.deleted_at > $3 or m.media_viewed_at > $3 or (m.media_expires_at > $3 and m.media_expires_at <= now()))`; const [clock, messages, removed, typing] = await Promise.all([ this.pool.query<{ cursor: Date }>(`select now() - interval '2 seconds' as cursor`), this.pool.query( `select ${columns} from messages m where m.match_id = $1 and ${visible}${since === null ? '' : ` and ${changed}`} order by m.created_at desc, m.id desc limit 500`, since === null ? [matchId, userId] : [matchId, userId, since], ), since === null ? null : this.pool.query<{ message_id: string }>( `select d.message_id from message_deletions d join messages m on m.id = d.message_id where m.match_id = $1 and d.user_id = $2 and d.deleted_at > $3`, [matchId, userId, since], ), options.includeTyping ? this.partnerTyping(userId, matchId) : null, ]); return { messages: messages.rows.map(toMessage), removed_ids: removed?.rows.map((row) => row.message_id) ?? [], server_time: clock.rows[0]!.cursor.toISOString(), complete: since === null, ...(typing ? { typing } : {}), }; } async sendText(userId: string, matchId: string, content: string): Promise { await this.assertMatchAccess(userId, matchId); const result = await this.pool.query( `insert into messages (match_id, user_id, content, version) values ($1, $2, $3, 1) returning id, match_id, user_id, content, created_at, read_at, delivered_at, deleted_at, media_path, media_type, media_expires_at, media_expired, media_viewed_at, version, encrypted_content, encryption_iv, keys_metadata, moderation_status, flag_reason, category`, [matchId, userId, content], ); return toMessage(result.rows[0]!); } async unreadCounts(userId: string): Promise<{ total: number; by_match: Record }> { const result = await this.pool.query<{ match_id: string; count: string }>( `select m.match_id, count(*)::text as count from messages m join matches ma on ma.id = m.match_id join profiles p on p.couple_id = ma.couple_id where p.id = $1 and m.user_id <> $1 and m.read_at is null and m.deleted_at is null and not exists ( select 1 from message_deletions d where d.message_id = m.id and d.user_id = $1 ) group by m.match_id`, [userId], ); const by_match = Object.fromEntries(result.rows.map((row) => [row.match_id, Number(row.count)])); return { total: Object.values(by_match).reduce((sum, count) => sum + count, 0), by_match }; } async markMatchRead(userId: string, matchId: string): Promise<{ updated: number; read_at: string }> { await this.assertMatchAccess(userId, matchId); const readAt = new Date(); const result = await this.pool.query( `update messages set delivered_at = coalesce(delivered_at, $3), read_at = $3 where match_id = $1 and user_id <> $2 and read_at is null`, [matchId, userId, readAt], ); return { updated: result.rowCount ?? 0, read_at: readAt.toISOString() }; } async markDelivered(userId: string, messageId: string): Promise { const message = await this.accessibleMessage(userId, messageId); if (message.user_id === userId) throw new ChatError('invalid_recipient', 'Senders cannot mark their own message delivered', 403); const result = await this.pool.query( `update messages m set delivered_at = coalesce(m.delivered_at, now()) where m.id = $1 returning ${columns}`, [messageId], ); return toMessage(result.rows[0]!); } async deleteForSelf(userId: string, messageId: string): Promise { await this.accessibleMessage(userId, messageId); await this.pool.query( `insert into message_deletions (message_id, user_id) values ($1, $2) on conflict do nothing`, [messageId, userId], ); } async deleteForEveryone(userId: string, messageId: string): Promise { const message = await this.accessibleMessage(userId, messageId); if (message.user_id !== userId) throw new ChatError('not_message_author', 'Only the sender can delete a message for everyone', 403); const result = await this.pool.query( `update messages m set deleted_at = coalesce(m.deleted_at, now()) where m.id = $1 returning ${columns}`, [messageId], ); return toMessage(result.rows[0]!); } async report(userId: string, messageId: string, reason: ReportReason): Promise { const message = await this.accessibleMessage(userId, messageId); if (message.user_id === userId) throw new ChatError('cannot_report_own_message', 'You cannot report your own message', 403); try { await this.pool.query( `insert into message_reports (message_id, reporter_id, reason) values ($1, $2, $3)`, [messageId, userId, reason], ); } catch (cause) { if (typeof cause === 'object' && cause !== null && 'code' in cause && cause.code === '23505') { throw new ChatError('already_reported', 'You have already reported this message', 409); } throw cause; } } async setTyping(userId: string, matchId: string, ttlMs: number): Promise { await this.assertMatchAccess(userId, matchId); await this.pool.query( `insert into chat_typing_states (match_id, user_id, expires_at) values ($1, $2, now() + ($3 * interval '1 millisecond')) on conflict (match_id, user_id) do update set expires_at = excluded.expires_at`, [matchId, userId, ttlMs], ); } /** Access already asserted by the caller, so the message poll can fold this in for free. */ private async partnerTyping(userId: string, matchId: string): Promise { const result = await this.pool.query<{ expires_at: Date }>( `select expires_at from chat_typing_states where match_id = $1 and user_id <> $2 and expires_at > now() order by expires_at desc limit 1`, [matchId, userId], ); const expiresAt = result.rows[0]?.expires_at ?? null; return { typing: Boolean(expiresAt), expires_at: iso(expiresAt) }; } async getPartnerTyping(userId: string, matchId: string): Promise { await this.assertMatchAccess(userId, matchId); return this.partnerTyping(userId, matchId); } async close(): Promise { await closeResolvedPool(this.pool, this.ownsPool); } }