feat(logs): messages infinite scroll and live refresh (#2214)

## Context

Fixes NAN-998
Contributes to NAN-908

More or less the same than #2213
The main diff here is that we don't need to update the current page
state, so we use two cursor `cursorBefore` and `cursorAfter` the former
to live refresh (prepend) and the later to do infinite scroll (append).

## Changes

- **Add `cursorBefore` and `cursorAfter`** to searchMessages
Note: there is no concept of search_before in Elasticsearch, you just
reverse the sorting order and re-reverse the results to achieve it.

- Implement live refresh and infinite scroll in the UI


## Test

**(note the scrollbar changing position because we prepend, it's very
hard to manage unless we use our own scroll bar


https://github.com/NangoHQ/nango/assets/1637651/4c802379-7eb7-4234-b5c8-b2ca6dff8526

packages/logs/lib/models/messages.integration.test.ts

@@ -1,7 +1,7 @@
 import { describe, beforeAll, it, expect, vi } from 'vitest';
 import { deleteIndex, migrateMapping } from '../es/helpers.js';
-import type { ListOperations } from './messages.js';
-import { listOperations } from './messages.js';
+import type { ListMessages, ListOperations } from './messages.js';
+import { listMessages, listOperations } from './messages.js';
 import { afterEach } from 'node:test';
 import { logContextGetter } from './logContextGetter.js';
 import type { OperationRowInsert } from '@nangohq/types';
@@ -54,4 +54,67 @@ describe('model', () => {
             expect(list3.cursor).toBeNull();
         });
     });
+
+    describe('messages', () => {
+        it('should list nothing', async () => {
+            const list = await listMessages({ limit: 10, parentId: '1' });
+            expect(list).toStrictEqual<ListMessages>({
+                cursorAfter: null,
+                cursorBefore: null,
+                count: 0,
+                items: []
+            });
+        });
+
+        it('should paginate', async () => {
+            const ctx = await logContextGetter.create(operationPayload, { start: false, account, environment }, { logToConsole: false });
+            await ctx.info('1');
+            await ctx.info('2');
+            await ctx.info('3');
+            await ctx.info('4');
+
+            // Should list 2 rows
+            const list1 = await listMessages({ limit: 2, parentId: ctx.id });
+            expect(list1.count).toBe(4);
+            expect(list1.items).toHaveLength(2);
+            expect(list1.cursorBefore).toBeDefined();
+            expect(list1.cursorAfter).toBeDefined();
+
+            // After:  Should list 2 more rows
+            const list2 = await listMessages({ limit: 2, parentId: ctx.id, cursorAfter: list1.cursorAfter });
+            expect(list2.count).toBe(4);
+            expect(list2.items).toHaveLength(2);
+            expect(list2.cursorAfter).toBeDefined();
+            expect(list2.items[0]!.id).not.toEqual(list1.items[0]!.id);
+
+            // After: Empty results
+            // When we get the second operation, it's not possible to know if there are more after so we still need to return a cursor
+            const list3 = await listMessages({ limit: 1, parentId: ctx.id, cursorAfter: list2.cursorAfter });
+            expect(list3.count).toBe(4);
+            expect(list3.items).toHaveLength(0);
+            expect(list2.cursorAfter).toBeDefined();
+
+            // Before: Should list 0 rows before
+            const list4 = await listMessages({ limit: 2, parentId: ctx.id, cursorBefore: list1.cursorBefore });
+            expect(list4.count).toBe(4);
+            expect(list4.items).toHaveLength(0);
+            expect(list4.cursorBefore).toBeNull();
+            expect(list4.cursorAfter).toBeDefined();
+
+            // Insert a new row
+            await ctx.info('4');
+            await ctx.info('5');
+
+            // Before: Should list 1 rows before
+            const list5 = await listMessages({ limit: 2, parentId: ctx.id, cursorBefore: list1.cursorBefore });
+            expect(list5.count).toBe(6);
+            expect(list5.items).toHaveLength(2);
+            expect(list5.items[0]?.message).toBe('5');
+            expect(list5.items[1]?.message).toBe('4');
+            expect(list5.cursorBefore).toBeDefined();
+            expect(list5.cursorAfter).toBeDefined();
+            expect(list5.items[0]!.id).not.toEqual(list1.items[0]!.id);
+            expect(list5.cursorBefore).not.toEqual(list1.cursorBefore);
+        });
+    });
 });

packages/logs/lib/models/messages.ts

@@ -23,6 +23,8 @@ export interface ListOperations {
 export interface ListMessages {
     count: number;
     items: MessageRow[];
+    cursorAfter: string | null;
+    cursorBefore: string | null;
 }
 export interface ListFilters {
     items: { key: string; doc_count: number }[];
@@ -225,6 +227,8 @@ export async function listMessages(opts: {
     limit: number;
     states?: SearchOperationsState[] | undefined;
     search?: string | undefined;
+    cursorBefore?: string | null | undefined;
+    cursorAfter?: string | null | undefined;
 }): Promise<ListMessages> {
     const query: estypes.QueryDslQueryContainer = {
         bool: {
@@ -249,20 +253,50 @@ export async function listMessages(opts: {
         });
     }
 
+    // Sort and cursor
+    let cursor: any[] | undefined;
+    let sort: estypes.Sort = [{ createdAt: 'desc' }, { id: 'desc' }];
+    if (opts.cursorBefore) {
+        // search_before does not exists so we reverse the sort
+        // https://github.com/elastic/elasticsearch/issues/29449
+        cursor = parseCursor(opts.cursorBefore);
+        sort = [{ createdAt: 'asc' }, { id: 'asc' }];
+    } else if (opts.cursorAfter) {
+        cursor = opts.cursorAfter ? parseCursor(opts.cursorAfter) : undefined;
+    }
+
     const res = await client.search<MessageRow>({
         index: indexMessages.index,
         size: opts.limit,
-        sort: [{ createdAt: 'desc' }, 'id'],
+        sort,
         track_total_hits: true,
+        search_after: cursor,
         query
     });
     const hits = res.hits;
 
+    const total = typeof hits.total === 'object' ? hits.total.value : hits.hits.length;
+    const totalPage = hits.hits.length;
+    const items = hits.hits.map((hit) => {
+        return hit._source!;
+    });
+
+    if (opts.cursorBefore) {
+        // In case we set before we have to reverse the message since we inverted the sort
+        items.reverse();
+        return {
+            count: total,
+            items,
+            cursorBefore: totalPage > 0 ? createCursor(hits.hits[hits.hits.length - 1]!) : null,
+            cursorAfter: null
+        };
+    }
+
     return {
-        count: typeof hits.total === 'object' ? hits.total.value : hits.hits.length,
-        items: hits.hits.map((hit) => {
-            return hit._source!;
-        })
+        count: total,
+        items,
+        cursorBefore: totalPage > 0 ? createCursor(hits.hits[0]!) : null,
+        cursorAfter: totalPage > 0 && total > totalPage && totalPage >= opts.limit ? createCursor(hits.hits[hits.hits.length - 1]!) : null
     };
 }
 

packages/server/lib/controllers/v1/logs/searchMessages.integration.test.ts

@@ -79,9 +79,9 @@ describe('POST /logs/messages', () => {
 
         isSuccess(res.json);
         expect(res.res.status).toBe(200);
-        expect(res.json).toStrictEqual({
+        expect(res.json).toStrictEqual<typeof res.json>({
             data: [],
-            pagination: { total: 0 }
+            pagination: { total: 0, cursorAfter: null, cursorBefore: null }
         });
     });
 
@@ -137,7 +137,7 @@ describe('POST /logs/messages', () => {
                     userId: null
                 }
             ],
-            pagination: { total: 1 }
+            pagination: { total: 1, cursorBefore: expect.any(String), cursorAfter: null }
         });
     });
 

packages/server/lib/controllers/v1/logs/searchMessages.ts

@@ -7,12 +7,15 @@ import { requireEmptyQuery, zodErrorToHTTP } from '@nangohq/utils';
 const validation = z
     .object({
         operationId: operationIdRegex,
-        limit: z.number().optional().default(100),
-        search: z.string().optional(),
+        limit: z.number().max(500).optional().default(100),
+        search: z.string().max(100).optional(),
         states: z
             .array(z.enum(['all', 'waiting', 'running', 'success', 'failed', 'timeout', 'cancelled']))
+            .max(10)
             .optional()
-            .default(['all'])
+            .default(['all']),
+        cursorBefore: z.string().or(z.null()).optional(),
+        cursorAfter: z.string().or(z.null()).optional()
     })
     .strict();
 
@@ -59,11 +62,13 @@ export const searchMessages = asyncWrapper<SearchMessages>(async (req, res) => {
         parentId: body.operationId,
         limit: body.limit!,
         states: body.states,
-        search: body.search
+        search: body.search,
+        cursorBefore: body.cursorBefore,
+        cursorAfter: body.cursorAfter
     });
 
     res.status(200).send({
         data: rawOps.items,
-        pagination: { total: rawOps.count }
+        pagination: { total: rawOps.count, cursorBefore: rawOps.cursorBefore, cursorAfter: rawOps.cursorAfter }
     });
 });

packages/types/lib/logs/api.ts

@@ -48,10 +48,17 @@ export type SearchMessages = Endpoint<{
     Method: 'POST';
     Path: '/api/v1/logs/messages';
     Querystring: { env: string };
-    Body: { operationId: string; limit?: number; states?: SearchOperationsState[]; search?: string | undefined };
+    Body: {
+        operationId: string;
+        limit?: number;
+        states?: SearchOperationsState[];
+        search?: string | undefined;
+        cursorBefore?: string | null | undefined;
+        cursorAfter?: string | null | undefined;
+    };
     Success: {
         data: MessageRow[];
-        pagination: { total: number };
+        pagination: { total: number; cursorBefore: string | null; cursorAfter: string | null };
     };
 }>;
 export type SearchMessagesData = SearchMessages['Success']['data'][0];

packages/webapp/src/components/ui/Drawer.tsx

@@ -24,7 +24,10 @@ const DrawerContent = React.forwardRef<React.ElementRef<typeof DrawerPrimitive.C
             <DrawerOverlay />
             <DrawerPrimitive.Content
                 ref={ref}
-                className={cn('fixed inset-x-0 bottom-0 top-0 z-50 flex flex-col bg-active-gray text-white border-l border-l-border-gray-400', className)}
+                className={cn(
+                    'outline-none fixed inset-x-0 bottom-0 top-0 z-50 flex flex-col bg-active-gray text-white border-l border-l-border-gray-400',
+                    className
+                )}
                 {...props}
             >
                 {children}

packages/webapp/src/components/ui/Table.tsx

@@ -2,9 +2,7 @@ import { forwardRef } from 'react';
 import { cn } from '../../utils/utils';
 
 const Table = forwardRef<HTMLTableElement, React.HTMLAttributes<HTMLTableElement>>(({ className, ...props }, ref) => (
-    <div className="relative w-full">
-        <table ref={ref} className={cn('w-full caption-bottom text-sm border-separate border-spacing-0', className)} {...props} />
-    </div>
+    <table ref={ref} className={cn('w-full caption-bottom text-sm border-separate border-spacing-0', className)} {...props} />
 ));
 Table.displayName = 'Table';
 

packages/webapp/src/hooks/useLogs.tsx

@@ -92,44 +92,59 @@ export function useSearchMessages(env: string, body: SearchMessages['Body']) {
     const [loading, setLoading] = useState<boolean>(false);
     const [data, setData] = useState<SearchMessages['Success']>();
     const [error, setError] = useState<SearchMessages['Errors']>();
+    const signal = useRef<AbortController | null>();
+
+    async function manualFetch(opts: Pick<SearchMessages['Body'], 'cursorAfter' | 'cursorBefore'>) {
+        if (signal.current && !signal.current.signal.aborted) {
+            signal.current.abort();
+        }
 
-    async function fetchData() {
         setLoading(true);
+        signal.current = new AbortController();
         try {
             const res = await fetch(`/api/v1/logs/messages?env=${env}`, {
                 method: 'POST',
-                body: JSON.stringify(body),
-                headers: { 'Content-Type': 'application/json' }
+                body: JSON.stringify({ ...body, ...opts }),
+                headers: { 'Content-Type': 'application/json' },
+                signal: signal.current.signal
             });
             if (res.status !== 200) {
-                setData(undefined);
-                setError((await res.json()) as SearchMessages['Errors']);
-                return;
+                return { error: (await res.json()) as SearchMessages['Errors'] };
             }
 
-            setError(undefined);
-            setData((await res.json()) as SearchMessages['Success']);
+            return { res: (await res.json()) as SearchMessages['Success'] };
         } catch (err) {
-            setData(undefined);
-            setError(err as any);
+            if (err instanceof DOMException && err.ABORT_ERR) {
+                return;
+            }
+            return { error: err };
         } finally {
             setLoading(false);
         }
     }
 
-    useEffect(() => {
-        if (!loading) {
-            void fetchData();
+    async function fetchData(opts: Pick<SearchMessages['Body'], 'cursorAfter' | 'cursorBefore'>) {
+        const man = await manualFetch(opts);
+        if (!man) {
+            return;
+        }
+        if (man.error) {
+            setData(undefined);
+            setError(man.error as any);
+            return;
         }
-    }, [env, body.operationId, body.limit, body.states, body.search]);
 
-    function trigger() {
+        setError(undefined);
+        setData(man.res);
+    }
+
+    function trigger(opts: Pick<SearchMessages['Body'], 'cursorAfter' | 'cursorBefore'>) {
         if (!loading) {
-            void fetchData();
+            void fetchData(opts);
         }
     }
 
-    return { data, error, loading, trigger };
+    return { data, error, loading, trigger, manualFetch };
 }
 
 export function useSearchFilters(enabled: boolean, env: string, body: SearchFilters['Body']) {

packages/webapp/src/pages/Logs/ShowMessage.tsx

@@ -56,7 +56,7 @@ export const ShowMessage: React.FC<{ message: MessageRow }> = ({ message }) => {
             <div className="overflow-x-auto">
                 <h4 className="font-semibold text-sm mb-2">Payload</h4>
 
-                {message.meta ? (
+                {message.meta || message.error ? (
                     <div className="text-gray-400 text-sm bg-pure-black py-2">
                         <Prism
                             language="json"
@@ -66,7 +66,7 @@ export const ShowMessage: React.FC<{ message: MessageRow }> = ({ message }) => {
                                 return { code: { padding: '0', whiteSpace: 'pre-wrap' } };
                             }}
                         >
-                            {JSON.stringify({ error: message.error || undefined, output: message.meta || undefined }, null, 2)}
+                            {JSON.stringify({ error: message.error?.message || undefined, output: message.meta || undefined }, null, 2)}
                         </Prism>
                     </div>
                 ) : (

packages/webapp/src/pages/Logs/ShowOperation.tsx

@@ -72,7 +72,7 @@ export const ShowOperation: React.FC<{ operationId: string }> = ({ operationId }
     }
 
     return (
-        <div className="py-8 px-6 flex flex-col gap-5">
+        <div className="py-8 px-6 flex flex-col gap-5 h-screen">
             <header className="flex gap-2 flex-col border-b border-b-gray-400 pb-5">
                 <h3 className="text-xl font-semibold text-white ">Operation Details</h3>
                 <div className="flex gap-3 items-center">
@@ -161,7 +161,7 @@ export const ShowOperation: React.FC<{ operationId: string }> = ({ operationId }
                     <div className="text-gray-400 text-xs bg-pure-black py-4 px-4">No payload.</div>
                 )}
             </div>
-            <SearchInOperation operationId={operationId} live={isLive} />
+            <SearchInOperation operationId={operationId} isLive={isLive} />
         </div>
     );
 };

packages/webapp/src/pages/Logs/components/MessageRow.tsx

@@ -12,9 +12,11 @@ export const MessageRow: React.FC<{ row: Row<SearchOperationsData> }> = ({ row }
     return (
         <Drawer direction="right" snapPoints={[drawerWidth]} handleOnly={true} noBodyStyles={true} nested>
             <DrawerTrigger asChild type={null as unknown as 'button'}>
-                <Table.Row data-state={row.getIsSelected() && 'selected'} className="hover:cursor-pointer border-b-border-gray-400">
+                <Table.Row data-state={row.getIsSelected() && 'selected'} className="hover:cursor-pointer border-b-border-gray-400 table table-fixed w-full ">
                     {row.getVisibleCells().map((cell) => (
-                        <Table.Cell key={cell.id}>{flexRender(cell.column.columnDef.cell, cell.getContext())}</Table.Cell>
+                        <Table.Cell key={cell.id} style={{ width: cell.column.columnDef.size }}>
+                            {flexRender(cell.column.columnDef.cell, cell.getContext())}
+                        </Table.Cell>
                     ))}
                 </Table.Row>
             </DrawerTrigger>

packages/webapp/src/pages/Logs/components/OperationRow.tsx

@@ -19,7 +19,7 @@ export const OperationRow: React.FC<{ row: Row<SearchOperationsData> }> = ({ row
                 </Table.Row>
             </DrawerTrigger>
             <DrawerContent>
-                <div className={`w-[1034px] relative select-text`}>
+                <div className={`w-[1034px] relative h-screen select-text`}>
                     <div className="absolute right-4 top-7">
                         <DrawerClose title="Close" className="w-8 h-8 flex items-center justify-center text-gray-400 hover:text-white">
                             <Cross1Icon className="" />