-
Notifications
You must be signed in to change notification settings - Fork 3.6k
feat(copilot): add copilot_chat_messages table with dual-write rollout #4724
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Closed
Closed
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| import { db } from '@sim/db' | ||
| import { createLogger } from '@sim/logger' | ||
| import { getErrorMessage } from '@sim/utils/errors' | ||
| import { sql } from 'drizzle-orm' | ||
|
|
||
| const logger = createLogger('CopilotChatMessagesCatchup') | ||
|
|
||
| /** | ||
| * Sweep recently-active chats from `copilot_chats.messages` JSONB into the new | ||
| * `copilot_chat_messages` table. Idempotent via `ON CONFLICT DO NOTHING`. | ||
| * | ||
| * Bounded to the last 7 days of activity so the cost is bounded regardless of | ||
| * total table size. The migration handles initial backfill; this sweep only | ||
| * exists to close the rolling-deploy window where old code may write to JSONB | ||
| * before the new dual-write code is live on every server. | ||
| */ | ||
| export async function catchUpCopilotChatMessages(): Promise<void> { | ||
| try { | ||
| const result = await db.execute(sql` | ||
| INSERT INTO copilot_chat_messages (chat_id, message_id, role, content, model, created_at, updated_at) | ||
| SELECT | ||
| c.id, | ||
| COALESCE(msg.value->>'id', gen_random_uuid()::text), | ||
|
waleedlatif1 marked this conversation as resolved.
|
||
| COALESCE(msg.value->>'role', 'user'), | ||
| msg.value, | ||
| COALESCE(msg.value->>'model', c.model), | ||
| COALESCE( | ||
| NULLIF(msg.value->>'createdAt','')::timestamp, | ||
| c.created_at + (msg.ord * interval '1 microsecond') | ||
| ), | ||
| COALESCE( | ||
| NULLIF(msg.value->>'createdAt','')::timestamp, | ||
| c.created_at + (msg.ord * interval '1 microsecond') | ||
| ) | ||
| FROM copilot_chats c | ||
| CROSS JOIN LATERAL jsonb_array_elements(c.messages) WITH ORDINALITY AS msg(value, ord) | ||
| WHERE c.updated_at > now() - interval '7 days' | ||
| AND jsonb_typeof(c.messages) = 'array' | ||
| AND jsonb_array_length(c.messages) > 0 | ||
| ON CONFLICT (chat_id, message_id) DO NOTHING | ||
| `) | ||
| logger.info('Copilot chat messages catch-up completed', { | ||
| rowCount: (result as { rowCount?: number }).rowCount ?? 0, | ||
| }) | ||
| } catch (err) { | ||
| logger.warn('Copilot chat messages catch-up failed', { error: getErrorMessage(err) }) | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| import { db } from '@sim/db' | ||
| import { copilotChatMessages } from '@sim/db/schema' | ||
| import { createLogger } from '@sim/logger' | ||
| import { getErrorMessage } from '@sim/utils/errors' | ||
| import { generateShortId } from '@sim/utils/id' | ||
| import { eq, sql } from 'drizzle-orm' | ||
|
|
||
| const logger = createLogger('CopilotChatMessagesDualWrite') | ||
|
|
||
| /** | ||
| * Build a row payload for `copilot_chat_messages` from a JSONB message blob. | ||
| * The blob format mirrors what's stored in the legacy `copilot_chats.messages` | ||
| * array — `id`, `role`, and optionally `model`/`createdAt` at the top level — | ||
| * with the entire blob preserved as the row's `content` for forward compat. | ||
| */ | ||
| function toMessageRow( | ||
| chatId: string, | ||
| rawMessage: unknown, | ||
| options?: { chatModel?: string | null; streamId?: string | null } | ||
| ): typeof copilotChatMessages.$inferInsert | null { | ||
| if (!rawMessage || typeof rawMessage !== 'object') return null | ||
| const msg = rawMessage as Record<string, unknown> | ||
| const id = typeof msg.id === 'string' && msg.id.length > 0 ? msg.id : generateShortId() | ||
| const role = typeof msg.role === 'string' ? msg.role : 'user' | ||
| const model = | ||
| typeof msg.model === 'string' && msg.model.length > 0 ? msg.model : (options?.chatModel ?? null) | ||
| return { | ||
| chatId, | ||
| messageId: id, | ||
| role, | ||
| content: msg, | ||
| model, | ||
| streamId: options?.streamId ?? null, | ||
| } | ||
|
waleedlatif1 marked this conversation as resolved.
|
||
| } | ||
|
|
||
| /** | ||
| * Append messages to the new `copilot_chat_messages` table. Best-effort — | ||
| * errors are logged but never thrown, since the legacy `copilot_chats.messages` | ||
| * JSONB column remains the source of truth during the dual-write rollout. | ||
| */ | ||
| export async function appendCopilotChatMessages( | ||
| chatId: string, | ||
| messages: unknown[], | ||
| options?: { chatModel?: string | null; streamId?: string | null } | ||
| ): Promise<void> { | ||
| if (!Array.isArray(messages) || messages.length === 0) return | ||
|
|
||
| const rows = messages | ||
| .map((msg) => toMessageRow(chatId, msg, options)) | ||
| .filter((row): row is typeof copilotChatMessages.$inferInsert => row !== null) | ||
|
|
||
| if (rows.length === 0) return | ||
|
|
||
| try { | ||
| await db | ||
| .insert(copilotChatMessages) | ||
| .values(rows) | ||
| .onConflictDoUpdate({ | ||
| target: [copilotChatMessages.chatId, copilotChatMessages.messageId], | ||
| set: { | ||
| content: sql`excluded.content`, | ||
| role: sql`excluded.role`, | ||
| // Preserve existing non-null model if the incoming row lacks one. | ||
| model: sql`COALESCE(excluded.model, ${copilotChatMessages.model})`, | ||
| updatedAt: sql`now()`, | ||
| }, | ||
|
waleedlatif1 marked this conversation as resolved.
|
||
| }) | ||
| } catch (err) { | ||
| logger.warn('Failed to append copilot chat messages', { | ||
| chatId, | ||
| messageCount: rows.length, | ||
| error: getErrorMessage(err), | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Replace all messages for a chat. Used by the update-messages endpoint that | ||
| * receives a full snapshot of the conversation state. Best-effort. | ||
| */ | ||
| export async function replaceCopilotChatMessages( | ||
| chatId: string, | ||
| messages: unknown[], | ||
| options?: { chatModel?: string | null } | ||
| ): Promise<void> { | ||
| if (!Array.isArray(messages)) return | ||
|
|
||
| const rows = messages | ||
| .map((msg) => toMessageRow(chatId, msg, options)) | ||
| .filter((row): row is typeof copilotChatMessages.$inferInsert => row !== null) | ||
|
|
||
| try { | ||
| await db.transaction(async (tx) => { | ||
| await tx.delete(copilotChatMessages).where(eq(copilotChatMessages.chatId, chatId)) | ||
| if (rows.length > 0) { | ||
| await tx.insert(copilotChatMessages).values(rows) | ||
| } | ||
| }) | ||
|
waleedlatif1 marked this conversation as resolved.
|
||
| } catch (err) { | ||
| logger.warn('Failed to replace copilot chat messages', { | ||
| chatId, | ||
| messageCount: rows.length, | ||
| error: getErrorMessage(err), | ||
| }) | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| CREATE TABLE "copilot_chat_messages" ( | ||
| "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, | ||
| "chat_id" uuid NOT NULL, | ||
| "message_id" text NOT NULL, | ||
| "role" text NOT NULL, | ||
| "content" jsonb NOT NULL, | ||
| "stream_id" text, | ||
| "parent_message_id" text, | ||
| "model" text, | ||
| "tokens_in" integer, | ||
| "tokens_out" integer, | ||
| "deleted_at" timestamp, | ||
| "created_at" timestamp DEFAULT now() NOT NULL, | ||
| "updated_at" timestamp DEFAULT now() NOT NULL | ||
| ); | ||
| --> statement-breakpoint | ||
| ALTER TABLE "copilot_chat_messages" ADD CONSTRAINT "copilot_chat_messages_chat_id_copilot_chats_id_fk" FOREIGN KEY ("chat_id") REFERENCES "public"."copilot_chats"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint | ||
| CREATE UNIQUE INDEX "copilot_chat_messages_chat_message_unique" ON "copilot_chat_messages" USING btree ("chat_id","message_id");--> statement-breakpoint | ||
| CREATE INDEX "copilot_chat_messages_chat_created_at_idx" ON "copilot_chat_messages" USING btree ("chat_id","created_at","id") WHERE "copilot_chat_messages"."deleted_at" IS NULL;--> statement-breakpoint | ||
| CREATE INDEX "copilot_chat_messages_chat_stream_idx" ON "copilot_chat_messages" USING btree ("chat_id","stream_id") WHERE "copilot_chat_messages"."stream_id" IS NOT NULL;--> statement-breakpoint | ||
| INSERT INTO "copilot_chat_messages" ( | ||
| "chat_id", "message_id", "role", "content", "model", "created_at", "updated_at" | ||
| ) | ||
| SELECT | ||
| c."id", | ||
| COALESCE(msg.value->>'id', gen_random_uuid()::text), | ||
| COALESCE(msg.value->>'role', 'user'), | ||
| msg.value, | ||
| COALESCE(msg.value->>'model', c."model"), | ||
| COALESCE( | ||
| NULLIF(msg.value->>'createdAt','')::timestamp, | ||
| c."created_at" + (msg.ord * interval '1 microsecond') | ||
| ), | ||
| COALESCE( | ||
| NULLIF(msg.value->>'createdAt','')::timestamp, | ||
| c."created_at" + (msg.ord * interval '1 microsecond') | ||
| ) | ||
| FROM "copilot_chats" c | ||
| CROSS JOIN LATERAL jsonb_array_elements(c."messages") WITH ORDINALITY AS msg(value, ord) | ||
| WHERE jsonb_typeof(c."messages") = 'array' | ||
| AND jsonb_array_length(c."messages") > 0 | ||
| ON CONFLICT ("chat_id", "message_id") DO NOTHING; |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.