byGalax
c9fe4879e0
Task 12 (the AuthContext userKeyState refactor) replaced the per-device DeviceRecord lookup with a localStorage UUID via ensureInstallId(). That UUID was then passed straight through to messages.sender_device_id on INSERT. The messages_insert_member RLS policy requires sender_device_id to be NULL OR to match a row in `devices` owned by the caller. The localStorage UUID matches neither -> 403 -> outbox endlessly retries with "Wiederhole". Fix: SendMessageParams.senderDeviceId becomes optional, and the message INSERT coerces undefined to NULL. The column is pure telemetry post-conv- keys so passing NULL is correct. Existing call sites that hand in ensureInstallId() still typecheck (string is assignable to string|null|undefined) but the row is written with NULL until those callers stop passing it.
542 lines
18 KiB
TypeScript
542 lines
18 KiB
TypeScript
import { bytesToUtf8, utf8ToBytes } from '../crypto/index';
|
|
import { bytesToPgHex, pgHexToBytes } from '../supabase/bytea';
|
|
import type { AppSupabaseClient } from '../supabase/client';
|
|
import {
|
|
decryptWithConvKey,
|
|
encryptWithConvKey,
|
|
getOrCreateConvKey,
|
|
type OwnUserCtx,
|
|
tryGetConvKey,
|
|
} from './convKeys';
|
|
import type { ChatMessage, DecryptedMessage } from './types';
|
|
|
|
const MESSAGE_COLS =
|
|
'id, conversation_id, sender_id, sender_device_id, reply_to_id, edited_at, deleted_at, created_at, ciphertext, nonce, key_version';
|
|
|
|
interface MessageRow {
|
|
id: string;
|
|
conversation_id: string;
|
|
sender_id: string;
|
|
sender_device_id: string | null;
|
|
reply_to_id: string | null;
|
|
edited_at: string | null;
|
|
deleted_at: string | null;
|
|
created_at: string;
|
|
ciphertext: string;
|
|
nonce: string;
|
|
key_version: number;
|
|
}
|
|
|
|
interface MessageWithCipher extends ChatMessage {
|
|
ciphertext: Uint8Array;
|
|
nonce: Uint8Array;
|
|
keyVersion: number;
|
|
}
|
|
|
|
function mapMessage(row: MessageRow): MessageWithCipher {
|
|
return {
|
|
id: row.id,
|
|
conversationId: row.conversation_id,
|
|
senderId: row.sender_id,
|
|
senderDeviceId: row.sender_device_id,
|
|
replyToId: row.reply_to_id,
|
|
editedAt: row.edited_at,
|
|
deletedAt: row.deleted_at,
|
|
createdAt: row.created_at,
|
|
ciphertext: pgHexToBytes(row.ciphertext),
|
|
nonce: pgHexToBytes(row.nonce),
|
|
keyVersion: row.key_version,
|
|
};
|
|
}
|
|
|
|
interface DeviceKey {
|
|
deviceId: string;
|
|
userId: string;
|
|
publicKey: Uint8Array;
|
|
}
|
|
|
|
// All device public keys for accepted members of a conversation.
|
|
export async function listConversationDeviceKeys(
|
|
client: AppSupabaseClient,
|
|
conversationId: string,
|
|
): Promise<DeviceKey[]> {
|
|
// 1. Members
|
|
const { data: members, error: mErr } = await client
|
|
.from('conversation_members')
|
|
.select('user_id, accepted')
|
|
.eq('conversation_id', conversationId);
|
|
if (mErr) throw mErr;
|
|
|
|
const memberIds = (members ?? []).filter((m) => m.accepted).map((m) => m.user_id);
|
|
if (memberIds.length === 0) return [];
|
|
|
|
// 2. Devices for those members
|
|
const { data: devices, error: dErr } = await client
|
|
.from('devices')
|
|
.select('id, user_id, public_key')
|
|
.in('user_id', memberIds);
|
|
if (dErr) throw dErr;
|
|
|
|
return (devices ?? [])
|
|
.filter((d): d is typeof d & { public_key: string } => d.public_key !== null)
|
|
.map((d) => ({
|
|
deviceId: d.id,
|
|
userId: d.user_id,
|
|
publicKey: pgHexToBytes(d.public_key),
|
|
}));
|
|
}
|
|
|
|
export interface SendMessageParams {
|
|
client: AppSupabaseClient;
|
|
conversationId: string;
|
|
plaintext: string;
|
|
senderUserId: string;
|
|
// Optional now: post-conv-keys this is pure telemetry. The 0.18 builds
|
|
// started passing a localStorage UUID that doesn't exist in the devices
|
|
// table; messages.sender_device_id RLS then 403'd every insert. Senders
|
|
// pass null (or an actually-registered device id, if they have one).
|
|
senderDeviceId?: string | null;
|
|
senderPrivateKey: Uint8Array;
|
|
replyToId?: string;
|
|
// Optional encrypted attachments — their handles are already materialised
|
|
// via `encryptAndUploadAttachment`. The caller is responsible for creating
|
|
// the corresponding message_attachments rows (see insertAttachmentRow) once
|
|
// the returned message id is known.
|
|
attachmentHandles?: import('./attachments.js').AttachmentHandle[];
|
|
}
|
|
|
|
// Encrypts and inserts a message using the shared per-conversation key
|
|
// (Sender-Key / Signal-style). The conv-key is generated lazily on first
|
|
// send and shared with every existing recipient device. New devices that
|
|
// register later receive their key bundle through `shareConvKeyToDevice`.
|
|
export async function sendEncryptedMessage(params: SendMessageParams): Promise<ChatMessage> {
|
|
const ownCtx: OwnUserCtx = {
|
|
userId: params.senderUserId,
|
|
privateKey: params.senderPrivateKey,
|
|
};
|
|
const handle = await getOrCreateConvKey(params.client, params.conversationId, ownCtx);
|
|
|
|
const attachments = params.attachmentHandles ?? [];
|
|
const payloadString =
|
|
attachments.length === 0
|
|
? params.plaintext
|
|
: JSON.stringify({ v: 1, text: params.plaintext, attachments });
|
|
const plainBytes = utf8ToBytes(payloadString);
|
|
|
|
const cipher = encryptWithConvKey(plainBytes, handle.key);
|
|
|
|
const insertPayload: Record<string, unknown> = {
|
|
conversation_id: params.conversationId,
|
|
sender_id: params.senderUserId,
|
|
// ALWAYS null until we re-introduce a real per-install devices row.
|
|
// Desktop callers currently pass a localStorage UUID (ensureInstallId)
|
|
// which doesn't exist in the devices table; the messages_insert_member
|
|
// RLS policy then 403s because the id can't be proven to belong to the
|
|
// caller. NULL satisfies the policy ("sender_device_id IS NULL OR …").
|
|
sender_device_id: null,
|
|
ciphertext: bytesToPgHex(cipher.ciphertext),
|
|
nonce: bytesToPgHex(cipher.nonce),
|
|
key_version: handle.keyVersion,
|
|
};
|
|
if (params.replyToId) insertPayload.reply_to_id = params.replyToId;
|
|
|
|
const { data: messageRow, error: insertErr } = await params.client
|
|
.from('messages')
|
|
.insert(insertPayload as never)
|
|
.select(MESSAGE_COLS)
|
|
.single();
|
|
if (insertErr) throw insertErr;
|
|
return mapMessage(messageRow as unknown as MessageRow);
|
|
}
|
|
|
|
// Fetch the last `limit` messages of a conversation in ascending order.
|
|
export async function fetchConversationMessages(
|
|
client: AppSupabaseClient,
|
|
conversationId: string,
|
|
limit = 100,
|
|
): Promise<MessageWithCipher[]> {
|
|
const { data, error } = await client
|
|
.from('messages')
|
|
.select(MESSAGE_COLS)
|
|
.eq('conversation_id', conversationId)
|
|
.order('created_at', { ascending: false })
|
|
.limit(limit);
|
|
if (error) throw error;
|
|
const rows = (data ?? []) as unknown as MessageRow[];
|
|
return rows.map(mapMessage).reverse();
|
|
}
|
|
|
|
export type { MessageWithCipher };
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Edit + delete
|
|
// ---------------------------------------------------------------------------
|
|
|
|
export interface EditMessageParams {
|
|
client: AppSupabaseClient;
|
|
messageId: string;
|
|
conversationId: string;
|
|
newPlaintext: string;
|
|
senderPrivateKey: Uint8Array;
|
|
}
|
|
|
|
// Re-encrypts the message body with the conv-key and updates the row.
|
|
// Server-side trigger enforces 24h window + sender-only rule.
|
|
export async function editEncryptedMessage(
|
|
params: EditMessageParams & { senderUserId: string; senderDeviceId?: string | null },
|
|
): Promise<void> {
|
|
const ownCtx: OwnUserCtx = {
|
|
userId: params.senderUserId,
|
|
privateKey: params.senderPrivateKey,
|
|
};
|
|
const handle = await getOrCreateConvKey(params.client, params.conversationId, ownCtx);
|
|
|
|
const cipher = encryptWithConvKey(utf8ToBytes(params.newPlaintext), handle.key);
|
|
|
|
const { error } = await params.client
|
|
.from('messages')
|
|
.update({
|
|
ciphertext: bytesToPgHex(cipher.ciphertext),
|
|
nonce: bytesToPgHex(cipher.nonce),
|
|
key_version: handle.keyVersion,
|
|
edited_at: new Date().toISOString(),
|
|
} as never)
|
|
.eq('id', params.messageId);
|
|
if (error) throw error;
|
|
}
|
|
|
|
export async function softDeleteMessage(
|
|
client: AppSupabaseClient,
|
|
messageId: string,
|
|
): Promise<void> {
|
|
const { error } = await client
|
|
.from('messages')
|
|
.update({ deleted_at: new Date().toISOString() } as never)
|
|
.eq('id', messageId);
|
|
if (error) throw error;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Read receipts
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// Upsert read rows for each message. Idempotent thanks to the composite PK.
|
|
export async function markMessagesRead(
|
|
client: AppSupabaseClient,
|
|
messageIds: string[],
|
|
): Promise<void> {
|
|
if (messageIds.length === 0) return;
|
|
const { data: session, error: aErr } = await client.auth.getUser();
|
|
if (aErr) throw aErr;
|
|
if (!session.user) return;
|
|
const myId = session.user.id;
|
|
const rows = messageIds.map((id) => ({ message_id: id, user_id: myId }));
|
|
const { error } = await client
|
|
.from('message_reads')
|
|
.upsert(rows as never, { onConflict: 'message_id,user_id', ignoreDuplicates: true });
|
|
if (error) throw error;
|
|
}
|
|
|
|
// Returns the set of message ids that `peerUserId` has read (among the given ids).
|
|
// RLS hides rows when either side has read receipts off.
|
|
export async function listPeerReadsForMessages(
|
|
client: AppSupabaseClient,
|
|
messageIds: string[],
|
|
peerUserId: string,
|
|
): Promise<Set<string>> {
|
|
if (messageIds.length === 0) return new Set();
|
|
const { data, error } = await client
|
|
.from('message_reads')
|
|
.select('message_id')
|
|
.eq('user_id', peerUserId)
|
|
.in('message_id', messageIds);
|
|
if (error) throw error;
|
|
return new Set((data ?? []).map((r) => r.message_id));
|
|
}
|
|
|
|
// Group variant: returns reads from ALL users (other than `excludeUserId`,
|
|
// typically caller themselves) keyed by message id → user id → timestamp.
|
|
// RLS already filters out users whose receipts are off.
|
|
export async function listGroupReadsForMessages(
|
|
client: AppSupabaseClient,
|
|
messageIds: string[],
|
|
excludeUserId: string,
|
|
): Promise<Map<string, Map<string, string>>> {
|
|
if (messageIds.length === 0) return new Map();
|
|
const { data, error } = await client
|
|
.from('message_reads')
|
|
.select('message_id, user_id, read_at')
|
|
.neq('user_id', excludeUserId)
|
|
.in('message_id', messageIds);
|
|
if (error) throw error;
|
|
const out = new Map<string, Map<string, string>>();
|
|
for (const row of data ?? []) {
|
|
const mid = row.message_id;
|
|
const uid = row.user_id;
|
|
const at = row.read_at;
|
|
let inner = out.get(mid);
|
|
if (!inner) {
|
|
inner = new Map();
|
|
out.set(mid, inner);
|
|
}
|
|
inner.set(uid, at);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
// Group variant of delivery receipts. Same shape as listGroupReadsForMessages.
|
|
export async function listGroupDeliveriesForMessages(
|
|
client: AppSupabaseClient,
|
|
messageIds: string[],
|
|
excludeUserId: string,
|
|
): Promise<Map<string, Map<string, string>>> {
|
|
if (messageIds.length === 0) return new Map();
|
|
const { data, error } = await (client as unknown as {
|
|
from: (t: string) => {
|
|
select: (cols: string) => {
|
|
neq: (col: string, val: string) => {
|
|
in: (
|
|
col: string,
|
|
vals: string[],
|
|
) => Promise<{
|
|
data:
|
|
| { message_id: string; user_id: string; delivered_at: string }[]
|
|
| null;
|
|
error: Error | null;
|
|
}>;
|
|
};
|
|
};
|
|
};
|
|
})
|
|
.from('message_deliveries')
|
|
.select('message_id, user_id, delivered_at')
|
|
.neq('user_id', excludeUserId)
|
|
.in('message_id', messageIds);
|
|
if (error) throw error;
|
|
const out = new Map<string, Map<string, string>>();
|
|
for (const row of data ?? []) {
|
|
let inner = out.get(row.message_id);
|
|
if (!inner) {
|
|
inner = new Map();
|
|
out.set(row.message_id, inner);
|
|
}
|
|
inner.set(row.user_id, row.delivered_at);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Delivery receipts
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// Records that the caller has received (fetched + decrypted) these messages.
|
|
// Idempotent; composite PK absorbs duplicates.
|
|
export async function markMessagesDelivered(
|
|
client: AppSupabaseClient,
|
|
messageIds: string[],
|
|
): Promise<void> {
|
|
if (messageIds.length === 0) return;
|
|
const { data: session, error: aErr } = await client.auth.getUser();
|
|
if (aErr) throw aErr;
|
|
if (!session.user) return;
|
|
const myId = session.user.id;
|
|
const rows = messageIds.map((id) => ({ message_id: id, user_id: myId }));
|
|
// `message_deliveries` is a later migration and not yet in the generated
|
|
// Supabase types; skip the strict table-name check for this call.
|
|
const { error } = await (client as unknown as {
|
|
from: (t: string) => {
|
|
upsert: (
|
|
rows: unknown,
|
|
opts: { onConflict: string; ignoreDuplicates: boolean },
|
|
) => Promise<{ error: Error | null }>;
|
|
};
|
|
})
|
|
.from('message_deliveries')
|
|
.upsert(rows, { onConflict: 'message_id,user_id', ignoreDuplicates: true });
|
|
if (error) throw error;
|
|
}
|
|
|
|
export async function listPeerDeliveriesForMessages(
|
|
client: AppSupabaseClient,
|
|
messageIds: string[],
|
|
peerUserId: string,
|
|
): Promise<Set<string>> {
|
|
if (messageIds.length === 0) return new Set();
|
|
const { data, error } = await (client as unknown as {
|
|
from: (t: string) => {
|
|
select: (cols: string) => {
|
|
eq: (col: string, val: string) => {
|
|
in: (
|
|
col: string,
|
|
vals: string[],
|
|
) => Promise<{ data: { message_id: string }[] | null; error: Error | null }>;
|
|
};
|
|
};
|
|
};
|
|
})
|
|
.from('message_deliveries')
|
|
.select('message_id')
|
|
.eq('user_id', peerUserId)
|
|
.in('message_id', messageIds);
|
|
if (error) throw error;
|
|
return new Set((data ?? []).map((r) => r.message_id));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Reactions
|
|
// ---------------------------------------------------------------------------
|
|
|
|
export interface MessageReaction {
|
|
messageId: string;
|
|
userId: string;
|
|
emoji: string;
|
|
createdAt: string;
|
|
}
|
|
|
|
export async function listReactionsForMessages(
|
|
client: AppSupabaseClient,
|
|
messageIds: string[],
|
|
): Promise<MessageReaction[]> {
|
|
if (messageIds.length === 0) return [];
|
|
const { data, error } = await client
|
|
.from('message_reactions')
|
|
.select('message_id, user_id, emoji, created_at')
|
|
.in('message_id', messageIds);
|
|
if (error) throw error;
|
|
return (data ?? []).map((r) => ({
|
|
messageId: r.message_id,
|
|
userId: r.user_id,
|
|
emoji: r.emoji,
|
|
createdAt: r.created_at,
|
|
}));
|
|
}
|
|
|
|
export async function addReaction(
|
|
client: AppSupabaseClient,
|
|
messageId: string,
|
|
emoji: string,
|
|
): Promise<void> {
|
|
const { data: session, error: aErr } = await client.auth.getUser();
|
|
if (aErr) throw aErr;
|
|
if (!session.user) throw new Error('not authenticated');
|
|
const { error } = await client.from('message_reactions').insert({
|
|
message_id: messageId,
|
|
user_id: session.user.id,
|
|
emoji,
|
|
} as never);
|
|
if (error) throw error;
|
|
}
|
|
|
|
export async function removeReaction(
|
|
client: AppSupabaseClient,
|
|
messageId: string,
|
|
emoji: string,
|
|
): Promise<void> {
|
|
const { data: session, error: aErr } = await client.auth.getUser();
|
|
if (aErr) throw aErr;
|
|
if (!session.user) throw new Error('not authenticated');
|
|
const { error } = await client
|
|
.from('message_reactions')
|
|
.delete()
|
|
.eq('message_id', messageId)
|
|
.eq('user_id', session.user.id)
|
|
.eq('emoji', emoji);
|
|
if (error) throw error;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Decrypt helpers
|
|
// ---------------------------------------------------------------------------
|
|
|
|
export interface DecryptParams {
|
|
client: AppSupabaseClient;
|
|
messages: MessageWithCipher[];
|
|
ownUserId: string;
|
|
ownPrivateKey: Uint8Array;
|
|
/**
|
|
* Optional delegate that performs the symmetric-decrypt + utf-8 decode
|
|
* step for a batch of messages. When provided, the main thread only does
|
|
* conv-key lookup; the CPU-heavy AEAD loop runs inside the delegate (e.g.
|
|
* a Web Worker). Items arrive with their per-message conv-key attached.
|
|
*/
|
|
aeadBatchDelegate?: (
|
|
items: Array<{
|
|
id: string;
|
|
ciphertext: Uint8Array;
|
|
nonce: Uint8Array;
|
|
key: Uint8Array;
|
|
}>,
|
|
) => Promise<Array<{ id: string; plaintext: string | null }>>;
|
|
}
|
|
|
|
// Decrypts messages using their conv-key (looked up + cached per
|
|
// keyVersion). Returns null `plaintext` when this device has no key bundle
|
|
// for that version yet (e.g. brand-new device waiting for share).
|
|
export async function decryptMessages(opts: DecryptParams): Promise<DecryptedMessage[]> {
|
|
const versions = new Map<string, Map<number, Uint8Array | null>>(); // convId -> version -> key | null
|
|
// First pass: resolve conv-keys for every message. Network-bound, stays on
|
|
// caller's thread so Supabase client + session remain usable.
|
|
interface Resolved {
|
|
message: MessageWithCipher;
|
|
key: Uint8Array | null;
|
|
}
|
|
const resolved: Resolved[] = [];
|
|
for (const m of opts.messages) {
|
|
let convCache = versions.get(m.conversationId);
|
|
if (!convCache) {
|
|
convCache = new Map();
|
|
versions.set(m.conversationId, convCache);
|
|
}
|
|
let key: Uint8Array | null;
|
|
if (convCache.has(m.keyVersion)) {
|
|
key = convCache.get(m.keyVersion) ?? null;
|
|
} else {
|
|
const handle = await tryGetConvKey(
|
|
opts.client,
|
|
m.conversationId,
|
|
opts.ownUserId,
|
|
opts.ownPrivateKey,
|
|
m.keyVersion,
|
|
);
|
|
key = handle?.key ?? null;
|
|
convCache.set(m.keyVersion, key);
|
|
}
|
|
resolved.push({ message: m, key });
|
|
}
|
|
|
|
// Second pass: symmetric decrypt. If a delegate is supplied (Web Worker),
|
|
// batch the CPU-heavy step to it; otherwise fall back to inline.
|
|
if (opts.aeadBatchDelegate) {
|
|
const batch = resolved
|
|
.filter((r): r is Resolved & { key: Uint8Array } => r.key !== null)
|
|
.map((r) => ({
|
|
id: r.message.id,
|
|
ciphertext: r.message.ciphertext,
|
|
nonce: r.message.nonce,
|
|
key: r.key,
|
|
}));
|
|
const plaintextById = new Map<string, string | null>();
|
|
if (batch.length > 0) {
|
|
const results = await opts.aeadBatchDelegate(batch);
|
|
for (const result of results) plaintextById.set(result.id, result.plaintext);
|
|
}
|
|
return resolved.map((r) => ({
|
|
...r.message,
|
|
plaintext: r.key ? plaintextById.get(r.message.id) ?? null : null,
|
|
}));
|
|
}
|
|
|
|
return resolved.map((r) => {
|
|
let plaintext: string | null = null;
|
|
if (r.key) {
|
|
try {
|
|
const decoded = decryptWithConvKey(r.message.ciphertext, r.message.nonce, r.key);
|
|
plaintext = bytesToUtf8(decoded);
|
|
} catch {
|
|
plaintext = null;
|
|
}
|
|
}
|
|
return { ...r.message, plaintext };
|
|
});
|
|
}
|