"use client";
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import Link from "next/link";
import { useSession } from "next-auth/react";
import { useChatSubscription } from "@/lib/use-chat-subscription";
import { useRouter } from "next/navigation";
import { useQueryClient } from "@tanstack/react-query";
import { Button } from "@/components/ui/button";
import { ScrollArea } from "@/components/ui/scroll-area";
import { ChatComposer } from "./chat-composer";
import { useChatMutationRecovery } from "./use-chat-mutation-recovery";
import { ChatMessageList, type ChatMessageView } from "./chat-message-list";
import {
advanceCatchUpWindow,
chatBadge,
createChatScope,
mergeChatMessages,
messageError,
rootReadIsProven,
type CatchUpWindow,
} from "./chat-utils";
import { api } from "@/lib/trpc";
const PAGE_SIZE = 50;
export function ReadFailureAlert({ onRetry }: { onRetry: () => void }) {
return (
Could not mark messages read. Unread status has not advanced.{" "}
);
}
type PendingMessage = {
key: string;
body: string;
parentMessageId?: string;
failed: boolean;
error?: string;
};
export function ChatWorkspace(props: { workspace: string; channelId?: string }) {
const userId = useSession().data?.user?.id;
return ;
}
function ChatWorkspaceScope({
workspace,
channelId,
}: {
workspace: string;
channelId?: string;
}) {
const router = useRouter();
const currentUserId = useSession().data?.user?.id;
const utils = api.useUtils();
const queryClient = useQueryClient();
const scopeRef = useRef(createChatScope(workspace, channelId));
// Render-time selection closes the gap before effects run after a route change.
scopeRef.current.select(workspace, channelId);
useEffect(() => {
const scope = createChatScope(workspace, channelId);
scopeRef.current = scope;
return () => scope.dispose();
}, [workspace, channelId]);
const revoked = scopeRef.current.isRevoked();
const purge = useCallback(() => {
if (scopeRef.current.isRevoked()) return;
scopeRef.current.revoke();
const valid = scopeRef.current.captureCleanup();
// Cancel in-flight reads before clearing cached chat data; stale responses
// must not repopulate a revoked conversation during navigation.
const chatQuery = {
predicate: (query: { queryKey: readonly unknown[] }) => {
const path = query.queryKey[0];
const options = query.queryKey[1];
return Array.isArray(path) && path[0] === "chat" &&
!!options && typeof options === "object" && "input" in options &&
!!options.input && typeof options.input === "object" &&
"workspace" in options.input && options.input.workspace === workspace;
},
};
void queryClient.cancelQueries(chatQuery).then(() => {
if (valid()) queryClient.removeQueries(chatQuery);
});
setRecent([]);
setPending([]);
setMessageChange(null);
setOlderRoots([]);
setLatestRoots([]);
setOlderReplies([]);
setLatestReplies([]);
setRefreshedRoot(null);
setThreadId(null);
router.replace(`/${workspace}/chat`);
}, [queryClient, router, workspace]);
const resolved = api.workspaces.resolve.useQuery({ handle: workspace });
const viewer = api.chat.viewer.useQuery({ workspace });
const channels = api.chat.listChannels.useQuery(
{ workspace },
{ refetchInterval: 15000 },
);
const summary = api.chat.unreadSummary.useQuery(
{ workspace },
{ refetchInterval: 15000 },
);
const channel = channels.data?.find((item) => item.id === channelId);
useEffect(() => {
if (channelId && channels.data && !channel && !channels.isError) purge();
}, [channelId, channels.data, channels.isError, channel, purge]);
const readable = !!channel && !!resolved.data && !revoked;
const recovery = useChatMutationRecovery({ workspace, channelId, enabled: readable });
const recoveryRefresh = recovery.refresh;
useChatSubscription(workspace, channels.data?.map((item) => item.id) ?? [], recoveryRefresh);
const denied = (
error: { data?: { code?: string } | null } | null | undefined,
) => error?.data?.code === "NOT_FOUND" || error?.data?.code === "FORBIDDEN";
const scope = { workspace, channelId: channelId ?? "" };
const [before, setBefore] = useState();
const [olderRoots, setOlderRoots] = useState([]);
const [latestRoots, setLatestRoots] = useState([]);
const history = api.chat.history.useQuery(
{ ...scope, limit: PAGE_SIZE, before },
{ enabled: readable },
);
const [threadId, setThreadId] = useState(null);
const threadScope = useRef(createChatScope(workspace, threadId ?? undefined));
threadScope.current.select(workspace, threadId ?? undefined);
const [threadBefore, setThreadBefore] = useState();
const [olderReplies, setOlderReplies] = useState([]);
const [latestReplies, setLatestReplies] = useState([]);
const [refreshedRoot, setRefreshedRoot] = useState(
null,
);
const thread = api.chat.threadHistory.useQuery(
{
...scope,
rootMessageId: threadId ?? "",
limit: PAGE_SIZE,
before: threadBefore,
},
{ enabled: readable && !!threadId },
);
const roots = useMemo(
() =>
mergeChatMessages(
mergeChatMessages(olderRoots, latestRoots),
history.data?.messages ?? [],
),
[olderRoots, latestRoots, history.data],
);
// There is no channel-wide high-water endpoint. Start from the oldest loaded
// root's predecessor so replies between loaded roots are never skipped.
// Empty channels still need zero to discover their first message.
const historyStart = history.data?.messages.reduce(
(min, item) =>
min === null || BigInt(item.sequence) < BigInt(min) ? item.sequence : min,
null,
);
const fallbackAfter = historyStart
? (BigInt(historyStart) - 1n).toString()
: "0";
const readBoundary =
summary.data?.channels.find((item) => item.channelId === channelId)
?.readSequence ?? "0";
// The visible page is a safe lower bound only when it is the entire root
// history and none of those roots has a reply that could lie before it.
const completeRootHistory =
history.data?.hasMore === false &&
history.data.messages.every((item) => item.replyCount === 0);
const initialAfter =
completeRootHistory || BigInt(readBoundary) >= BigInt(fallbackAfter)
? fallbackAfter
: readBoundary;
const [catchUpWindow, setCatchUpWindow] = useState(
null,
);
const activeWindow = catchUpWindow ?? { after: initialAfter };
const catchUp = api.chat.catchUp.useQuery(
{ ...scope, ...activeWindow, limit: 100 },
{
enabled: readable && !!history.data && !!summary.data,
refetchInterval: 5000,
},
);
const [scanned, setScanned] = useState(null);
const [unseenReply, setUnseenReply] = useState(null);
const [recent, setRecent] = useState([]);
const [pending, setPending] = useState([]);
const [messageChange, setMessageChange] = useState<{
message: ChatMessageView;
type: "edit_message" | "delete_message";
body: string;
key: string;
failed: boolean;
error?: string;
} | null>(null);
const editMutation = api.chat.editMessage.useMutation();
const deleteMutation = api.chat.deleteMessage.useMutation();
const accessRevoked = [
history.error,
catchUp.error,
thread.error,
summary.error,
viewer.error,
recovery.error as { data?: { code?: string } } | null,
].some(denied);
useEffect(() => {
if (accessRevoked) purge();
}, [accessRevoked, purge]);
const previousScope = useRef(`${workspace}:${channelId ?? ""}`);
useEffect(() => {
const nextScope = `${workspace}:${channelId ?? ""}`;
if (previousScope.current === nextScope) return;
previousScope.current = nextScope;
setRecent([]);
setPending([]);
setMessageChange(null);
processedPage.current = undefined;
setThreadId(null);
setBefore(undefined);
setOlderRoots([]);
setLatestRoots([]);
setRefreshedRoot(null);
setThreadBefore(undefined);
setOlderReplies([]);
setCatchUpWindow(null);
setScanned(null);
setUnseenReply(null);
}, [channelId, workspace]);
useEffect(() => {
setThreadBefore(undefined);
setOlderReplies([]);
setLatestReplies([]);
setRefreshedRoot(null);
}, [threadId]);
// Previously paged roots and replies are held outside React Query. Mirror refreshed
// cached pages into those arrays so edits to old pages become visible too.
useEffect(
() =>
queryClient.getQueryCache().subscribe((event) => {
if (
event.type !== "updated" ||
event.action.type !== "success" ||
scopeRef.current.isRevoked()
)
return;
const key = event.query.queryKey;
const path = key[0];
const options = key[1];
if (
!Array.isArray(path) ||
path[0] !== "chat" ||
(path[1] !== "history" && path[1] !== "threadHistory") ||
!options ||
typeof options !== "object" ||
!("input" in options)
)
return;
const input = options.input;
if (
!input ||
typeof input !== "object" ||
!("channelId" in input) ||
!("workspace" in input) ||
input.channelId !== channelId ||
input.workspace !== workspace
)
return;
const data: unknown = event.query.state.data;
if (
!data ||
typeof data !== "object" ||
!("messages" in data || "replies" in data)
)
return;
if (
path[1] === "history" &&
"messages" in data &&
Array.isArray(data.messages)
) {
const messages = data.messages as ChatMessageView[];
setOlderRoots((current) => mergeChatMessages(current, messages));
}
if (
path[1] === "threadHistory" &&
"replies" in data &&
Array.isArray(data.replies) &&
"rootMessageId" in input &&
input.rootMessageId === threadId
) {
setOlderReplies((current) =>
mergeChatMessages(current, data.replies as ChatMessageView[]),
);
if ("root" in data && data.root && typeof data.root === "object")
setRefreshedRoot((current) => mergeChatMessages(current ? [current] : [], [data.root as ChatMessageView])[0]);
}
}),
[queryClient, workspace, channelId, threadId],
);
const processedPage = useRef(undefined);
useEffect(() => {
const page = catchUp.data;
if (!page || scopeRef.current.isRevoked() || processedPage.current === page) return;
// A cached page from another interval must not advance this interval.
if (
BigInt(page.next) < BigInt(activeWindow.after) ||
(activeWindow.through !== undefined &&
page.through !== activeWindow.through)
)
return;
processedPage.current = page;
setRecent((current) => mergeChatMessages(current, page.messages));
if (page.messages.length) {
// Catch-up also carries durable revisions for messages already paged in.
setOlderRoots((current) =>
mergeChatMessages(
current,
page.messages.filter((item) => !item.parentMessageId),
),
);
setOlderReplies((current) =>
mergeChatMessages(
current,
page.messages.filter((item) => item.parentMessageId === threadId),
),
);
}
setScanned((current) =>
current &&
BigInt(current) > BigInt(page.hasMore ? page.next : page.through)
? current
: page.hasMore
? page.next
: page.through,
);
const reply = page.messages.find((item) => item.parentMessageId);
if (reply)
setUnseenReply((current) =>
current && BigInt(current) < BigInt(reply.sequence)
? current
: reply.sequence,
);
setCatchUpWindow((current) =>
advanceCatchUpWindow(
current ?? { after: activeWindow.after, through: activeWindow.through },
page,
),
);
}, [catchUp.data, activeWindow.after, activeWindow.through, threadId]);
const sendMutation = api.chat.send.useMutation();
const readMutation = api.chat.markChannelRead.useMutation({
onSuccess: () => {
if (!scopeRef.current.isRevoked()) void utils.chat.unreadSummary.invalidate({ workspace });
},
});
const reading = useRef(null);
const acknowledged = useRef("0");
const [readFailure, setReadFailure] = useState(null);
useEffect(() => {
reading.current = null;
acknowledged.current = "0";
setReadFailure(null);
}, [channelId, workspace]);
const viewed = useCallback(
(sequence: string) => {
if (
!readable ||
!channelId ||
document.visibilityState !== "visible" ||
readFailure ||
!rootReadIsProven(sequence, scanned, unseenReply, readBoundary) ||
reading.current
)
return;
if (
BigInt(sequence) <= BigInt(readBoundary) ||
BigInt(sequence) <= BigInt(acknowledged.current)
)
return;
const valid = scopeRef.current.capture();
reading.current = sequence;
readMutation.mutate(
{ workspace, channelId, sequence },
{
onSuccess: () => {
if (valid()) acknowledged.current = sequence;
},
onError: (error) => {
if (!valid()) return;
if (denied(error)) purge();
else setReadFailure(sequence);
},
onSettled: () => {
if (valid()) reading.current = null;
},
},
);
},
[
readBoundary,
readFailure,
scanned,
unseenReply,
readable,
channelId,
readMutation,
workspace,
purge,
],
);
const send = useCallback(
async (
body: string,
parentMessageId?: string,
key = crypto.randomUUID(),
) => {
if (!resolved.data || !channelId || !readable || channel?.archivedAt)
return;
const valid = scopeRef.current.capture();
const requestedThread = threadId;
const validThread = threadScope.current.capture();
setPending((items) => [
...items.filter((item) => item.key !== key),
{ key, body, parentMessageId, failed: false },
]);
try {
const result = await sendMutation.mutateAsync({
workspace,
command: {
type: "send_message",
workspaceId: resolved.data.id,
channelId,
context: { type: "workspace", workspaceId: resolved.data.id },
body,
clientIdempotencyKey: key,
...(parentMessageId ? { parentMessageId } : {}),
},
});
if (!valid()) return;
const confirmed: ChatMessageView = {
...result.message,
createdAt: new Date(),
};
if (parentMessageId && validThread()) {
setLatestReplies((current) =>
mergeChatMessages(current, [confirmed]),
);
} else if (!parentMessageId) {
setLatestRoots((current) => mergeChatMessages(current, [confirmed]));
}
setPending((items) => items.filter((item) => item.key !== key));
// Refreshes are best-effort after confirmation. A failed refresh must
// never turn a persisted send into a retryable pending row.
void Promise.allSettled([
utils.chat.history.invalidate({ workspace, channelId }),
utils.chat.history
.fetch({ workspace, channelId, limit: PAGE_SIZE })
.then((page) => {
if (!valid()) return;
setLatestRoots((current) =>
mergeChatMessages(current, page.messages),
);
}),
utils.chat.threadHistory.invalidate({ workspace, channelId }),
...(parentMessageId
? [
utils.chat.threadHistory
.fetch({
workspace,
channelId,
rootMessageId: parentMessageId,
limit: PAGE_SIZE,
})
.then((page) => {
if (valid() && validThread() && requestedThread === parentMessageId)
setLatestReplies((current) =>
mergeChatMessages(current, page.replies),
);
}),
]
: []),
utils.chat.catchUp.invalidate({ workspace, channelId }),
utils.chat.unreadSummary.invalidate({ workspace }),
]);
} catch (error) {
if (!valid()) return;
const code =
error &&
typeof error === "object" &&
"data" in error &&
error.data &&
typeof error.data === "object" &&
"code" in error.data
? String(error.data.code)
: undefined;
if (code === "NOT_FOUND" || code === "FORBIDDEN") {
purge();
} else if (!scopeRef.current.isRevoked()) {
setPending((items) =>
items.map((item) =>
item.key === key
? { ...item, failed: true, error: messageError(code) }
: item,
),
);
}
}
},
[
resolved.data,
channelId,
channel?.archivedAt,
readable,
sendMutation,
workspace,
utils,
purge,
threadId,
],
);
const applyMessageChange = useCallback(
async (change: NonNullable) => {
if (!resolved.data || !channelId || !readable || channel?.archivedAt)
return;
const valid = scopeRef.current.capture();
const { message, type, body, key } = change;
const command = {
workspaceId: resolved.data.id,
channelId,
context: { type: "workspace" as const, workspaceId: resolved.data.id },
messageId: message.id,
expectedRevision: message.revision ?? 1,
clientMutationId: key,
};
setMessageChange({ ...change, failed: false, error: undefined });
try {
const outcome =
type === "edit_message"
? await editMutation.mutateAsync({
workspace,
command: { ...command, type, body },
})
: await deleteMutation.mutateAsync({
workspace,
command: { ...command, type },
});
if (!valid()) return;
const updated: ChatMessageView = {
...message,
revision: outcome.result.revision,
status: outcome.result.status,
body: type === "delete_message" ? "" : body.trim(),
...(type === "delete_message"
? { deletedAt: new Date() }
: { editedAt: new Date() }),
};
setRecent((current) =>
current.some((item) => item.id === updated.id)
? mergeChatMessages(current, [updated])
: current,
);
setOlderRoots((current) =>
current.map((item) =>
item.id === updated.id
? mergeChatMessages([item], [updated])[0]
: item,
),
);
if ("parentMessageId" in message && message.parentMessageId) {
setLatestReplies((current) => mergeChatMessages(current, [updated]));
} else {
setLatestRoots((current) => mergeChatMessages(current, [updated]));
}
setOlderReplies((current) =>
current.map((item) =>
item.id === updated.id
? mergeChatMessages([item], [updated])[0]
: item,
),
);
setMessageChange(null);
// Old root and thread pages may be retained locally. Refetch every cached page,
// not only the newest page; revision-aware merging prevents late stale responses
// from resurrecting an older edit or a tombstone.
void Promise.allSettled([
queryClient.refetchQueries({
predicate: (query) => {
const path = query.queryKey[0];
const options = query.queryKey[1];
return (
Array.isArray(path) &&
path[0] === "chat" &&
(path[1] === "history" ||
path[1] === "threadHistory" ||
path[1] === "catchUp") &&
!!options && typeof options === "object" && "input" in options &&
!!options.input && typeof options.input === "object" &&
"workspace" in options.input && options.input.workspace === workspace
);
},
type: "all",
}),
utils.chat.unreadSummary.invalidate({ workspace }),
]);
} catch (error) {
if (!valid()) return;
if (denied(error as { data?: { code?: string } })) {
purge();
return;
}
const code =
error &&
typeof error === "object" &&
"data" in error &&
error.data &&
typeof error.data === "object" &&
"code" in error.data
? String(error.data.code)
: undefined;
if (code === "CONFLICT") {
void queryClient.refetchQueries({
predicate: (query) => {
const path = query.queryKey[0];
const options = query.queryKey[1];
return (
Array.isArray(path) &&
path[0] === "chat" &&
(path[1] === "history" || path[1] === "threadHistory") &&
!!options && typeof options === "object" && "input" in options &&
!!options.input && typeof options.input === "object" &&
"workspace" in options.input && options.input.workspace === workspace
);
},
type: "all",
});
}
setMessageChange((current) =>
current?.key === key
? {
...current,
failed: true,
error:
code === "CONFLICT"
? "Message changed. Refresh messages before editing again."
: messageError(code),
}
: current,
);
}
},
[
resolved.data,
channelId,
readable,
channel?.archivedAt,
editMutation,
deleteMutation,
workspace,
queryClient,
utils,
purge,
],
);
const recovered = (messages: ChatMessageView[]) =>
mergeChatMessages(messages, recovery.messages.flatMap((item) => {
const loaded = messages.find((message) => message.id === item.id);
return loaded ? [{ ...loaded, ...item }] : [];
}));
const rootsWithRecent = useMemo(
() =>
mergeChatMessages(
mergeChatMessages(roots, latestRoots),
recent.filter(
(item) => !("parentMessageId" in item) || !item.parentMessageId,
) as typeof roots,
),
[roots, latestRoots, recent],
);
const replies = useMemo(
() =>
mergeChatMessages(
mergeChatMessages(olderReplies, latestReplies),
thread.data?.replies ?? [],
),
[olderReplies, latestReplies, thread.data],
);
const visibleReplies = useMemo(
() =>
mergeChatMessages(
mergeChatMessages(replies, latestReplies),
recent.filter(
(item) =>
"parentMessageId" in item && item.parentMessageId === threadId,
) as typeof replies,
),
[replies, latestReplies, recent, threadId],
);
const canModerate =
viewer.data?.role === "admin" ||
viewer.data?.role === "owner" ||
channel?.ownerUserId === currentUserId;
const startChange = (
message: ChatMessageView,
type: "edit_message" | "delete_message",
) => {
if (
!message.revision ||
channel?.archivedAt ||
message.status === "deleted" ||
message.deletedAt ||
(type === "edit_message"
? message.authorUserId !== currentUserId
: message.authorUserId !== currentUserId && !canModerate)
)
return;
setMessageChange({
message,
type,
body: message.body,
key: crypto.randomUUID(),
failed: false,
});
};
const counts = new Map(
summary.data?.channels.map((item) => [item.channelId, item]),
);
return (
{!channelId ? (
Select a channel to start chatting.
) : channels.isLoading || resolved.isLoading ? (
Loading channel…
) : channels.isError || resolved.isError ? (
Could not load channel.{" "}
) : accessRevoked || !channel ? (
Channel unavailable or access removed.
) : (
{channel.name ?? "Direct message"}
{history.isLoading && (
Loading messages…
)}
{history.isError && !accessRevoked && (
Could not load messages.{" "}
)}
{history.data?.hasMore && (
)}
{!history.isLoading && !rootsWithRecent.length && (
No messages yet.
)}
startChange(message, "edit_message")}
onDelete={(message) => startChange(message, "delete_message")}
pending={pending
.filter((item) => !item.parentMessageId)
.map((item) => ({
...item,
retry: () => void send(item.body, undefined, item.key),
}))}
/>
{messageChange && (
{messageChange.type === "edit_message"
? "Edit message"
: "Delete message?"}
{messageChange.type === "edit_message" && (
)}
{readFailure && (
{
const sequence = readFailure;
setReadFailure(null);
reading.current = sequence;
readMutation.mutate(
{ workspace, channelId, sequence },
{
onSuccess: () => {
acknowledged.current = sequence;
},
onError: (error) => {
if (denied(error)) purge();
else setReadFailure(sequence);
},
onSettled: () => {
reading.current = null;
},
},
);
}}
/>
)}
{catchUp.isError && !accessRevoked && (
Could not check for new messages.{" "}
)}
{!channel.archivedAt && (
void send(body)}
label="Message channel"
/>
)}
)}
{channel && threadId && !accessRevoked && (
)}
);
}