Agents: fix joining (broken create_direct_conversation), full rollback (0.7.1)

package.json

@@ -1,7 +1,7 @@
 {
   "name": "qryptchat-web",
   "private": true,
-  "version": "0.7.0",
+  "version": "0.7.1",
   "type": "module",
   "bin": {
     "qc": "./bin/qc.js",

src/lib/agents/agents.js

@@ -118,8 +118,30 @@ export async function redeemInvite(db, token, { username, displayName, publicKey
 	if (!claimed?.length) throw new AgentError('invite_invalid', 'This invite was just used', 410);
 
 	let authUserId = null;
+	let userId = null;
+	let conversationId = null;
+	// Everything made here is removed again on failure: deleting the auth user
+	// does not cascade to public.users, and a trigger gives every new user a
+	// note-to-self conversation.
 	const undo = async () => {
-		if (authUserId) await db.auth.admin.deleteUser(authUserId).catch(() => {});
+		const quietly = (p) => Promise.resolve(p).catch(() => {});
+		if (conversationId) {
+			await quietly(db.from('conversation_participants').delete().eq('conversation_id', conversationId));
+			await quietly(db.from('conversations').delete().eq('id', conversationId));
+		}
+		if (userId) {
+			const { data: own } = await db.from('conversations').select('id').eq('created_by', userId).then((r) => r, () => ({ data: [] }));
+			for (const c of own ?? []) {
+				await quietly(db.from('conversation_participants').delete().eq('conversation_id', c.id));
+				await quietly(db.from('conversations').delete().eq('id', c.id));
+			}
+			await quietly(db.from('conversation_participants').delete().eq('user_id', userId));
+			await quietly(db.from('users').delete().eq('id', userId));
+		}
+		if (authUserId) {
+			await quietly(db.from('user_public_keys').delete().eq('user_id', authUserId));
+			await quietly(db.auth.admin.deleteUser(authUserId));
+		}
 		await db.from('agent_invites').update({ redeemed_at: null, redeemed_by: null }).eq('id', invite.id);
 	};
 
@@ -148,17 +170,30 @@ export async function redeemInvite(db, token, { username, displayName, publicKey
 			if (userError?.code === '23505') throw new AgentError('username_taken', 'That username is taken', 409);
 			throw new AgentError('server_error', 'Could not create the agent account', 500);
 		}
+		userId = user.id;
 
 		const { error: keyError } = await db
 			.from('user_public_keys')
 			.insert({ user_id: authUserId, public_key: keyBytes.toString('base64'), key_type: 'ML-KEM-1024' });
 		if (keyError) throw new AgentError('server_error', 'Could not store the agent key', 500);
 
-		const { data: conversationId, error: convError } = await db.rpc('create_direct_conversation', {
-			user1_id: invite.inviter_user_id,
-			user2_id: user.id,
-		});
-		if (convError || !conversationId) throw new AgentError('server_error', 'Could not open the conversation', 500);
+		// The direct conversation with the operator. (create_direct_conversation()
+		// collides with the trigger that already adds the creator as a participant.)
+		const { data: conv, error: convError } = await db
+			.from('conversations')
+			.insert({ type: 'direct', created_by: invite.inviter_user_id })
+			.select('id')
+			.single();
+		if (convError || !conv?.id) throw new AgentError('server_error', 'Could not open the conversation', 500);
+		conversationId = conv.id;
+		const { error: partError } = await db.from('conversation_participants').upsert(
+			[
+				{ conversation_id: conversationId, user_id: invite.inviter_user_id },
+				{ conversation_id: conversationId, user_id: user.id },
+			],
+			{ onConflict: 'conversation_id,user_id', ignoreDuplicates: true },
+		);
+		if (partError) throw new AgentError('server_error', 'Could not open the conversation', 500);
 
 		await db.from('agent_invites').update({ redeemed_by: user.id }).eq('id', invite.id);
 

src/lib/agents/agents.test.js

@@ -8,7 +8,7 @@ const KEY = Buffer.alloc(1568, 7).toString('base64');
 
 /** A tiny in-memory stand-in for the service-role client. */
 function fakeDb({ invite, usernameTaken = false, keyInsertFails = false } = {}) {
-	const state = { invite: invite ? { ...invite } : null, users: [], keys: [], deletedAuth: [], rpc: [] };
+	const state = { invite: invite ? { ...invite } : null, users: [], keys: [], deletedAuth: [], conversations: [], participants: [], deleted: [] };
 	const db = {
 		state,
 		auth: {
@@ -17,7 +17,7 @@ function fakeDb({ invite, usernameTaken = false, keyInsertFails = false } = {})
 				deleteUser: vi.fn(async (id) => state.deletedAuth.push(id))
 			}
 		},
-		rpc: vi.fn(async (name, args) => (state.rpc.push([name, args]), { data: 'conv-1', error: null })),
+		rpc: vi.fn(async () => ({ data: null, error: { message: 'create_direct_conversation must not be used' } })),
 		from(table) {
 			let op = 'select';
 			let patch = null;
@@ -29,11 +29,28 @@ function fakeDb({ invite, usernameTaken = false, keyInsertFails = false } = {})
 				ilike: (k, v) => (filters.push(['ilike', k, v]), q),
 				update: (p) => ((op = 'update'), (patch = p), q),
 				insert: (row) => ((op = 'insert'), (patch = row), q),
+				upsert: (rows) => ((op = 'upsert'), (patch = rows), q),
+				delete: () => ((op = 'delete'), q),
 				maybeSingle: async () => run('one'),
 				single: async () => run('one'),
 				then: (res, rej) => run('many').then(res, rej)
 			};
 			async function run(shape) {
+				if (op === 'delete') {
+					state.deleted.push([table, ...filters.map(([, k, v]) => `${k}=${v}`)]);
+					return { data: null, error: null };
+				}
+				if (table === 'conversations') {
+					if (op === 'insert') {
+						state.conversations.push(patch);
+						return { data: { id: 'conv-1' }, error: null };
+					}
+					return { data: [], error: null };
+				}
+				if (table === 'conversation_participants') {
+					state.participants.push(...[].concat(patch ?? []));
+					return { data: null, error: null };
+				}
 				if (table === 'agent_invites') {
 					const inv = state.invite;
 					const match = inv && filters.every(([t, k, v]) => (t === 'is' ? inv[k] === v || (v === null && inv[k] == null) : k === 'token_hash' ? inv.token_hash === v : inv[k] === v));
@@ -116,7 +133,9 @@ describe('redeemInvite', () => {
 		expect(out.conversationId).toBe('conv-1');
 		expect(db.state.users[0]).toMatchObject({ account_type: 'agent', operator_user_id: 'inviter', username: 'athena_bot', display_name: 'Athena' });
 		expect(db.state.keys[0]).toEqual({ user_id: 'auth-agent', public_key: KEY, key_type: 'ML-KEM-1024' });
-		expect(db.rpc).toHaveBeenCalledWith('create_direct_conversation', { user1_id: 'inviter', user2_id: 'user-agent' });
+		expect(db.rpc).not.toHaveBeenCalled();
+		expect(db.state.conversations[0]).toEqual({ type: 'direct', created_by: 'inviter' });
+		expect(db.state.participants.map((p) => p.user_id)).toEqual(['inviter', 'user-agent']);
 		expect(db.state.invite.redeemed_at).toBeTruthy();
 		expect(db.state.invite.redeemed_by).toBe('user-agent');
 	});
@@ -136,6 +155,8 @@ describe('redeemInvite', () => {
 		const db = fakeDb({ invite: openInviteFor(token), keyInsertFails: true });
 		await expect(redeemInvite(db, token, { username: 'athena_bot', publicKey: KEY })).rejects.toBeInstanceOf(AgentError);
 		expect(db.state.deletedAuth).toEqual(['auth-agent']);
+		// The users row goes too: deleting the auth user does not cascade to public.users.
+		expect(db.state.deleted).toContainEqual(['users', 'id=user-agent']);
 		expect(db.state.invite.redeemed_at).toBeNull();
 	});
 });