diff --git a/apps/api/src/routes/messages/first-neutral-send.integration.test.ts b/apps/api/src/routes/messages/first-neutral-send.integration.test.ts new file mode 100644 index 00000000..fc79b309 --- /dev/null +++ b/apps/api/src/routes/messages/first-neutral-send.integration.test.ts @@ -0,0 +1,204 @@ +import { expect, test } from "bun:test"; +import { + db, + reconcileChannelSpineConcurrentIndexes, +} from "@wateaminbox/database"; +import { DEFAULT_SLA_WEEKLY_SCHEDULE } from "@wateaminbox/shared"; +import { sql } from "kysely"; +import { app } from "../../app.js"; +import { hashPassword } from "../../lib/password.js"; +import { ensureActiveCaseWithin } from "../../services/conversation-case.service.js"; +import { + clearTenantConnection, + createTenantSchema, + getSchemaName, + getTenantConnection, +} from "../../services/tenant.service.js"; + +const integration = process.env.RUN_DB_INTEGRATION === "1" ? test : test.skip; + +const PASSWORD = "Correct-Horse-123!"; + +/** + * The first outgoing message to a contact that has never had one. + * + * The "new chat" flow inserts a `contacts` row and nothing else, and the + * channel-spine bridge is otherwise only written as a shadow of a legacy + * message - so under neutral write authority the send had no conversation to + * write against and answered 404. Every later message to the same contact + * worked, which is what made it read as a contact problem rather than a + * bridge one. The route has to build the bridge itself. + */ +integration( + "sends the first message to a contact that has no conversation yet", + async () => { + const companyId = crypto.randomUUID(); + const schemaName = getSchemaName(companyId); + const ownerId = crypto.randomUUID(); + const ownerEmail = `first-send-${ownerId}@example.com`; + const connectionId = crypto.randomUUID(); + const contactId = crypto.randomUUID(); + try { + await db + .insertInto("users") + .values({ + id: ownerId, + email: ownerEmail, + password_hash: await hashPassword(PASSWORD), + email_verified_at: new Date(), + }) + .execute(); + await db + .insertInto("companies") + .values({ + id: companyId, + name: "First neutral send", + schema_name: schemaName, + status: "active", + }) + .execute(); + await db + .insertInto("company_members") + .values({ company_id: companyId, user_id: ownerId, role: "owner" }) + .execute(); + await db + .insertInto("sla_policies") + .values({ + company_id: companyId, + target_minutes: 60, + direct_resolution_target_minutes: 480, + group_response_target_minutes: 120, + group_resolution_target_minutes: 960, + timezone: "UTC", + weekly_schedule: JSON.stringify(DEFAULT_SLA_WEEKLY_SCHEDULE), + exceptions: JSON.stringify([]), + effective_from: new Date("1970-01-01T00:00:00Z"), + created_by: ownerId, + }) + .execute(); + await db + .insertInto("channel_spine_workspace_flags") + .values({ + company_id: companyId, + dual_write_enabled: true, + dual_write_revision: "test", + shadow_normalization_enabled: true, + shadow_normalization_revision: "test", + neutral_reads_enabled: true, + neutral_read_revision: "test", + write_authority: "neutral", + write_authority_revision: "test", + enabled_providers: sql< + string[] + >`ARRAY['whatsapp_linked_device']::text[]`, + provider_enable_revision: "test", + revision: "1", + created_by: ownerId, + updated_by: ownerId, + }) + .execute(); + await createTenantSchema(companyId); + await reconcileChannelSpineConcurrentIndexes(db, schemaName); + const tenantDb = await getTenantConnection(companyId); + + await tenantDb + .insertInto("whatsapp_connections") + .values({ + id: connectionId, + name: "MMPhone", + phone_number: "60123456789", + jid: "60123456789@s.whatsapp.net", + status: "connected", + }) + .execute(); + await tenantDb + .insertInto("whatsapp_connection_sessions") + .values({ whatsapp_connection_id: connectionId, status: "connected" }) + .execute(); + // Exactly what the outbound "new chat" flow leaves behind: a contact + // row, no conversation, no bridge. + await tenantDb + .insertInto("contacts") + .values({ + id: contactId, + whatsapp_connection_id: connectionId, + jid: "33786137909@s.whatsapp.net", + phone_number: "33786137909", + }) + .execute(); + + // The chat is open, as it is in the inbox before anyone types: the send + // guard requires an active case, and this test is about the bridge. + await tenantDb.transaction().execute((trx) => + ensureActiveCaseWithin( + trx, + { id: contactId, isGroup: false }, + { + companyId, + openedBy: ownerId, + reason: "Outbound conversation started in the inbox", + }, + ), + ); + + const login = await app.request("/api/auth/login", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ email: ownerEmail, password: PASSWORD }), + }); + expect(login.status).toBe(200); + const { tokens } = (await login.json()) as { + tokens: { accessToken: string }; + }; + + const response = await app.request("/api/messages", { + method: "POST", + headers: { + authorization: `Bearer ${tokens.accessToken}`, + "x-company-id": companyId, + "content-type": "application/json", + }, + body: JSON.stringify({ + contactId, + content: "first message", + messageType: "text", + }), + }); + expect(response.status).toBe(200); + + // The bridge the send had to build, and the message written against it. + const conversation = await tenantDb + .selectFrom("conversations") + .select(["id", "channel_account_id"]) + .where("legacy_contact_id", "=", contactId) + .executeTakeFirstOrThrow(); + expect(conversation.channel_account_id).toBe(connectionId); + const message = await tenantDb + .selectFrom("messages") + .select(["content", "contact_id"]) + .executeTakeFirstOrThrow(); + expect(message.content).toBe("first message"); + expect(message.contact_id).toBe(contactId); + } finally { + clearTenantConnection(companyId); + await sql + .raw(`DROP SCHEMA IF EXISTS "${schemaName}" CASCADE`) + .execute(db); + await db + .deleteFrom("channel_spine_workspace_flags") + .where("company_id", "=", companyId) + .execute(); + await db + .deleteFrom("sla_policies") + .where("company_id", "=", companyId) + .execute(); + await db + .deleteFrom("company_members") + .where("company_id", "=", companyId) + .execute(); + await db.deleteFrom("companies").where("id", "=", companyId).execute(); + await db.deleteFrom("users").where("id", "=", ownerId).execute(); + } + }, + 60_000, +); diff --git a/apps/api/src/routes/messages/send.ts b/apps/api/src/routes/messages/send.ts index e1b43b43..1cd8906e 100644 --- a/apps/api/src/routes/messages/send.ts +++ b/apps/api/src/routes/messages/send.ts @@ -5,12 +5,16 @@ */ import { zValidator } from "../../lib/validator.js"; +import type { TenantDatabase } from "@wateaminbox/database"; import { toDbDate } from "@wateaminbox/shared"; import type { Context } from "hono"; import { Hono } from "hono"; -import { sql } from "kysely"; +import { type Kysely, sql } from "kysely"; import type { z } from "zod"; -import { shadowLinkedDeviceLegacyMutation } from "../../channel-spine/providers/whatsapp-linked-device/shadow.js"; +import { + shadowLinkedDeviceLegacyMutation, + shadowLinkedDeviceWorkflow, +} from "../../channel-spine/providers/whatsapp-linked-device/shadow.js"; import { badRequest, notFound } from "../../lib/errors.js"; import { buildOutboundMediaColumns } from "../../lib/message-formatters.js"; import { isConfirmedQuote } from "../../lib/message-quote.js"; @@ -72,6 +76,36 @@ const messageSendRateLimiter = createConditionalRateLimiter( type SendMessageBody = z.infer; +/** + * The conversation for a contact, building the bridge first when the contact + * does not have one yet. + * + * The bridge is otherwise only ever written as a shadow of a legacy message, + * so a contact that has never exchanged a message has no conversation row - + * which is exactly the state the outbound "new chat" flow leaves behind, since + * it inserts nothing but the `contacts` row. Under neutral write authority the + * send needs that conversation, and refusing it here made the first outgoing + * message to a new contact impossible while every later one succeeded. + * + * Unlike `shadowLinkedDeviceLegacyMutation`, this is not best-effort: the send + * depends on the conversation, so a bridge that cannot be resolved fails the + * request instead of being journalled. Every identity in the bridge is + * deterministic and every write upserts, so two first sends racing each other + * converge on the same conversation. + */ +async function ensureChannelConversationId( + tenantDb: Kysely, + companyId: string, + contactId: string, +): Promise { + const existing = await conversationIdForContact(tenantDb, contactId); + if (existing) return existing; + return tenantDb.transaction().execute(async (trx) => { + const bridged = await shadowLinkedDeviceWorkflow(trx, companyId, contactId); + return bridged.status === "ready" ? bridged.bridge.conversationId : null; + }); +} + async function sendChannelContactMessage( c: Context, contact: { id: string }, @@ -79,7 +113,11 @@ async function sendChannelContactMessage( mentionedJids?: string[], ) { const { tenantDb, user, companyId } = getRouteContext(c); - const conversationId = await conversationIdForContact(tenantDb, contact.id); + const conversationId = await ensureChannelConversationId( + tenantDb, + companyId, + contact.id, + ); if (!conversationId) return notFound(c, "Contact or JID"); const conversation = await tenantDb .selectFrom("conversations")