diff --git a/__tests__/unit/services/cat-worker-bootstrap.test.ts b/__tests__/unit/services/cat-worker-bootstrap.test.ts new file mode 100644 index 000000000..6ae7a6e30 --- /dev/null +++ b/__tests__/unit/services/cat-worker-bootstrap.test.ts @@ -0,0 +1,84 @@ +/** + * The Cat has to be able to come into existence. + * + * There is a circularity hiding in this feature, and the first version shipped + * into it: `resolveMentions` can only flag @cat when a Cat PROFILE exists, so + * with no account nothing is ever queued — and if the account were only + * established when something was queued, nothing ever would be. A freshly + * deployed platform would sit there with @cat resolving to nobody, looking + * exactly like a working feature nobody had used yet. + * + * So the worker establishes the account BEFORE it looks at the queue, and + * before the empty-queue exit. This test is the thing that stops that ordering + * from being "tidied up" later. + */ + +const ensureCatAccount = jest.fn(); +const claimCatMentions = jest.fn(); + +jest.mock('@/services/mentions/cat-account', () => ({ + ensureCatAccount: (...a: unknown[]) => ensureCatAccount(...a), +})); +jest.mock('@/services/mentions/queue', () => ({ + claimCatMentions: (...a: unknown[]) => claimCatMentions(...a), + completeCatMention: jest.fn(), + failCatMention: jest.fn(), + MAX_ATTEMPTS: 3, +})); +jest.mock('@/services/mentions/cat-reply', () => ({ + replyToConversationMention: jest.fn().mockResolvedValue(true), +})); + +import { runCatMentions } from '@/services/mentions/worker'; + +beforeEach(() => { + ensureCatAccount.mockReset().mockResolvedValue({ id: 'cat-1', username: 'cat' }); + claimCatMentions.mockReset().mockResolvedValue([]); +}); + +describe('the mention worker bootstraps the Cat', () => { + it('establishes the account even when the queue is empty', async () => { + await runCatMentions({} as never); + // The empty queue is the NORMAL state, and it is exactly the state a new + // deployment is in. Returning early without this call is the deadlock. + expect(ensureCatAccount).toHaveBeenCalledTimes(1); + }); + + it('establishes the account before it claims anything', async () => { + const order: string[] = []; + ensureCatAccount.mockImplementation(async () => { + order.push('ensure'); + return { id: 'cat-1', username: 'cat' }; + }); + claimCatMentions.mockImplementation(async () => { + order.push('claim'); + return []; + }); + + await runCatMentions({} as never); + expect(order).toEqual(['ensure', 'claim']); + }); + + it('still reports an empty run as empty', async () => { + await expect(runCatMentions({} as never)).resolves.toEqual({ + claimed: 0, + answered: 0, + failed: 0, + }); + }); + + it('answers a claimed mention once the account exists', async () => { + claimCatMentions.mockResolvedValue([ + { id: 'q1', source_type: 'message', source_id: 'm1', requester_id: 'u1', conversation_id: 'c1', parent_event_id: null, attempts: 1 }, + ]); + await expect(runCatMentions({} as never)).resolves.toMatchObject({ claimed: 1, answered: 1 }); + }); + + it('fails claimed mentions rather than speaking as nobody', async () => { + ensureCatAccount.mockResolvedValue(null); + claimCatMentions.mockResolvedValue([ + { id: 'q1', source_type: 'message', source_id: 'm1', requester_id: 'u1', conversation_id: 'c1', parent_event_id: null, attempts: 1 }, + ]); + await expect(runCatMentions({} as never)).resolves.toMatchObject({ failed: 1, answered: 0 }); + }); +}); diff --git a/scripts/systemd/orangecat-cron@cat-account.timer b/scripts/systemd/orangecat-cron@cat-account.timer index c7a481a15..59149a287 100644 --- a/scripts/systemd/orangecat-cron@cat-account.timer +++ b/scripts/systemd/orangecat-cron@cat-account.timer @@ -9,6 +9,12 @@ Description=OrangeCat Cat account invariant (daily) [Timer] +# OnActiveSec is what makes a freshly deployed Cat exist within the minute. +# `Persistent=true` only replays a run MISSED while the machine was down; a +# timer enabled for the first time simply waits for the next OnCalendar, which +# for a daily timer is up to 24 hours away. That gap is not cosmetic — until the +# account exists, @cat resolves to nobody and nothing can be queued. +OnActiveSec=1min OnCalendar=daily Persistent=true RandomizedDelaySec=15m diff --git a/src/features/messaging/server/mutations.ts b/src/features/messaging/server/mutations.ts index 63d3e8dfd..e8785fb64 100644 --- a/src/features/messaging/server/mutations.ts +++ b/src/features/messaging/server/mutations.ts @@ -1,7 +1,7 @@ import { callRpc, fromTable } from '@/lib/supabase/untyped'; import { createAdminClient } from '@/lib/supabase/admin'; +import { writeMessage } from './write-message'; import { logger } from '@/utils/logger'; -import type { Json } from '@/types/database'; import { DATABASE_TABLES } from '@/config/database-tables'; import { createConvRecord, @@ -9,8 +9,6 @@ import { getServerUser, type ConversationParticipantsInsert, type ConversationParticipantsUpdate, - type ConversationsUpdate, - type MessagesInsert, type ProfilesInsert, } from './shared'; @@ -85,66 +83,20 @@ export async function sendMessage( }); } - // Insert the message directly using admin client to bypass RLS - const messageData: MessagesInsert = { - conversation_id: conversationId, - sender_id: user.id, - content: content, - message_type: type, - metadata: (metadata || {}) as Json, - }; - - // Add sender_actor_id if provided (column added by migration) - if (senderActorId) { - (messageData as MessagesInsert & { sender_actor_id?: string }).sender_actor_id = - senderActorId; - } - - const { data: message, error: insertError } = await fromTable(admin, DATABASE_TABLES.MESSAGES) - .insert(messageData) - .select('id') - .single(); - - if (insertError || !message) { - logger.error('Error inserting message:', insertError); - throw Object.assign(new Error('Failed to send message'), { status: 500 }); - } - - // Update conversation metadata using admin client - const conversationUpdate: ConversationsUpdate = { - last_message_at: new Date().toISOString(), - last_message_preview: content.substring(0, 100), - last_message_sender_id: senderId, - updated_at: new Date().toISOString(), - }; - - const { error: updateError } = await fromTable(admin, DATABASE_TABLES.CONVERSATIONS) - .update(conversationUpdate) - .eq('id', conversationId); - - if (updateError) { - logger.warn('Failed to update conversation metadata:', updateError); - // Don't fail the message send for this - } - - // Update participant's last_read_at for sender using admin client to avoid RLS recursion - const participantUpdate: ConversationParticipantsUpdate = { - last_read_at: new Date().toISOString(), - }; - - const updateQuery = fromTable(admin, DATABASE_TABLES.CONVERSATION_PARTICIPANTS) - .update(participantUpdate) - .eq('conversation_id', conversationId) - .eq('user_id', senderId); - const { error: readError } = await updateQuery; - - if (readError) { - logger.warn('Failed to update sender read time:', readError); - // Don't fail the message send for this - } - - logger.info('Message sent successfully:', message.id); - return message.id; + // Authorization is done; the writing half lives in write-message.ts so a + // sender without a browser session (the Cat, answering from a worker) can + // reuse it without this function's user checks being relaxed for everyone. + const messageId = await writeMessage(admin, { + conversationId, + senderId: user.id, + content, + type, + metadata, + senderActorId, + }); + + logger.info('Message sent successfully:', messageId); + return messageId; } catch (error) { logger.error('Error sending message:', error); throw error; diff --git a/src/features/messaging/server/write-message.ts b/src/features/messaging/server/write-message.ts new file mode 100644 index 000000000..0d2e83baf --- /dev/null +++ b/src/features/messaging/server/write-message.ts @@ -0,0 +1,96 @@ +/** + * Persisting a message, with authorization already decided elsewhere. + * + * Extracted from `sendMessage`, which does two separable things: it decides + * whether the caller may speak as `senderId` — requiring a signed-in user whose + * id matches — and then it writes. Those are different concerns, and conflating + * them meant there was no way for a sender that is not a browser session to + * write at all. + * + * The Cat is exactly that sender. It answers from a cron worker with no session, + * so calling `sendMessage` threw `Unauthorized` and its reply was lost — found + * by running the real path against production, not by any unit test, because a + * mocked client has no opinion about sessions. + * + * The fix is NOT to relax `sendMessage`. Its "sender must match the + * authenticated user" check is a real control on the human path and stays + * exactly as it was; it now delegates the writing half to this. A caller + * reaching this function is asserting that authorization has already happened — + * which is why it takes an admin client explicitly rather than making one. + */ + +import { fromTable } from '@/lib/supabase/untyped'; +import { DATABASE_TABLES } from '@/config/database-tables'; +import { logger } from '@/utils/logger'; +import type { Json } from '@/types/database'; +import type { SupabaseClient } from '@supabase/supabase-js'; + +export interface WriteMessageInput { + conversationId: string; + senderId: string; + content: string; + type?: string; + metadata?: Record | null; + senderActorId?: string | null; +} + +/** + * @returns the new message id. + * @throws when the message itself cannot be stored. The two follow-up writes + * are best-effort: a conversation whose preview is stale is a cosmetic + * problem, while a message that was never written is a lost one. + */ +export async function writeMessage( + admin: SupabaseClient, + input: WriteMessageInput +): Promise { + const { conversationId, senderId, content } = input; + + const messageData: Record = { + conversation_id: conversationId, + sender_id: senderId, + content, + message_type: input.type ?? 'text', + metadata: (input.metadata || {}) as Json, + }; + if (input.senderActorId) { + messageData.sender_actor_id = input.senderActorId; + } + + const { data: message, error: insertError } = await fromTable(admin, DATABASE_TABLES.MESSAGES) + .insert(messageData) + .select('id') + .single(); + + if (insertError || !message) { + logger.error('Error inserting message:', insertError); + throw Object.assign(new Error('Failed to send message'), { status: 500 }); + } + + // Without this the conversation list still shows the previous message as the + // latest, so a reply that was written looks like it never arrived. + const { error: updateError } = await fromTable(admin, DATABASE_TABLES.CONVERSATIONS) + .update({ + last_message_at: new Date().toISOString(), + last_message_preview: content.substring(0, 100), + last_message_sender_id: senderId, + updated_at: new Date().toISOString(), + }) + .eq('id', conversationId); + + if (updateError) { + logger.warn('Failed to update conversation metadata:', updateError); + } + + // Your own message is read by definition. + const { error: readError } = await fromTable(admin, DATABASE_TABLES.CONVERSATION_PARTICIPANTS) + .update({ last_read_at: new Date().toISOString() }) + .eq('conversation_id', conversationId) + .eq('user_id', senderId); + + if (readError) { + logger.warn('Failed to update sender read time:', readError); + } + + return message.id as string; +} diff --git a/src/services/mentions/cat-reply.ts b/src/services/mentions/cat-reply.ts index fa8f7c9e9..cc4029345 100644 --- a/src/services/mentions/cat-reply.ts +++ b/src/services/mentions/cat-reply.ts @@ -19,7 +19,7 @@ import { } from '@/config/cat-identity'; import { DATABASE_TABLES } from '@/config/database-tables'; import { callPlatformJson, parseJsonLoose } from '@/services/cat/platform-llm'; -import { sendMessage } from '@/features/messaging/server/mutations'; +import { writeMessage } from '@/features/messaging/server/write-message'; import { logger } from '@/utils/logger'; import type { SupabaseClient } from '@supabase/supabase-js'; @@ -121,7 +121,15 @@ export async function replyToConversationMention( } await ensureCatIsParticipant(admin, conversationId, catId); - await sendMessage(conversationId, catId, reply, 'text', { is_cat_reply: true }); + // writeMessage, not sendMessage: the latter requires a signed-in user whose id + // matches the sender, which is right for a person and impossible for a worker. + // Going through it threw Unauthorized and the reply was silently lost. + await writeMessage(admin, { + conversationId, + senderId: catId, + content: reply, + metadata: { is_cat_reply: true }, + }); return true; } diff --git a/src/services/mentions/worker.ts b/src/services/mentions/worker.ts index 698dac4ce..1c6e02b86 100644 --- a/src/services/mentions/worker.ts +++ b/src/services/mentions/worker.ts @@ -33,15 +33,21 @@ export async function runCatMentions( ): Promise { const result: MentionRunResult = { claimed: 0, answered: 0, failed: 0 }; + // BEFORE claiming, and before the empty-queue exit, because otherwise the + // system deadlocks on itself: resolveMentions can only flag @cat when a Cat + // profile exists, so with no account nothing ever queues — and if the account + // were only created when something was queued, nothing ever would be. The + // every-minute tick is therefore also what brings the Cat into existence. + // Cheap enough to do unconditionally: one primary-key lookup when it is a + // no-op, which is always after the first run. + const cat = await ensureCatAccount(admin); + const claimed = await claimCatMentions(admin, limit); result.claimed = claimed.length; if (claimed.length === 0) { return result; } - // Established once per tick rather than per mention: it is one indexed lookup - // when it is a no-op, and without it there is no sender to speak as. - const cat = await ensureCatAccount(admin); if (!cat) { for (const mention of claimed) { await failCatMention(admin, mention, 'no Cat account');