feat(import-app): stream import

Makefile

@@ -38,4 +38,4 @@ add-component-compliance-files: ## Add the compliance files into all the compone
 
 .PHONY: codeclean
 codeclean: ## Code Clean
-	@yarn prettier --write .
+	@pnpm run prettier --write .

components/crystallize-import/package.json

@@ -31,17 +31,18 @@
         "react-icons": "^5.0.1",
         "react-tiny-popover": "^8.0.4",
         "read-excel-file": "^5.7.1",
+        "remix-utils": "^7.5.0",
         "uuid": "^9.0.1"
     },
     "devDependencies": {
         "@remix-run/dev": "^2.6.0",
         "@types/d3": "^7.4.3",
         "@types/jsonwebtoken": "^9.0.5",
         "@types/react": "^18.2.55",
-        "autoprefixer": "^10.4.17",
         "@types/react-datepicker": "^4.19.6",
         "@types/react-dom": "^18.2.19",
         "@types/uuid": "^9.0.8",
+        "autoprefixer": "^10.4.17",
         "postcss-import": "^16.0.0",
         "prettier": "^3.2.5",
         "tailwindcss": "^3.4.1",

components/crystallize-import/src/contracts/form-submission.ts

@@ -1,4 +1,5 @@
 export interface FormSubmission {
+    importId?: string;
     shapeIdentifier: string;
     folderPath: string;
     rows: Record<string, any>[];

components/crystallize-import/src/contracts/ui-types.ts

@@ -89,6 +89,7 @@ export type State = {
     loading?: boolean;
     done?: boolean;
     channels: Record<string, string[]>;
+    importId: string;
     preflight?: {
         validCount: number;
         errorCount: number;

components/crystallize-import/src/core.server/import-runner.server.ts

@@ -1,7 +1,13 @@
 import { Bootstrapper, EVENT_NAMES, JsonSpec } from '@crystallize/import-utilities';
-import { error } from 'ajv/dist/vocabularies/applicator/dependencies';
+import { EventEmitter } from 'events';
+import util from 'util';
+
+export const dump = (obj: any, depth?: number) => {
+    console.log(util.inspect(obj, false, depth, true));
+};
 
 type Deps = {
+    emitter: EventEmitter;
     tenantIdentifier: string;
     sessionId: string | undefined;
     skipPublication?: boolean;
@@ -24,10 +30,12 @@ type Subscriptons = {
 };
 
 export const runImport = async (
+    importUuid: string,
     spec: JsonSpec,
     { onItemCreated, onItemUpdated }: Subscriptons,
-    { tenantIdentifier, sessionId, skipPublication, verbose }: Deps,
+    { tenantIdentifier, sessionId, skipPublication, verbose, emitter }: Deps,
 ) => {
+    dump({ spec }, 200);
     return new Promise((resolve) => {
         const errors: any = [];
         const bootstrapper = new Bootstrapper();
@@ -50,13 +58,27 @@ export const runImport = async (
         }
 
         if (onItemCreated) {
-            bootstrapper.on(EVENT_NAMES.ITEM_CREATED, (data) => onItemCreated(data).catch((err) => errors.push(err)));
+            bootstrapper.on(EVENT_NAMES.ITEM_CREATED, (data) => {
+                emitter.emit(importUuid, {
+                    event: 'item-created',
+                    data,
+                });
+                onItemCreated(data).catch((err) => errors.push(err));
+            });
         }
         if (onItemUpdated) {
-            bootstrapper.on(EVENT_NAMES.ITEM_UPDATED, (data) => onItemUpdated(data).catch((err) => errors.push(err)));
+            bootstrapper.on(EVENT_NAMES.ITEM_UPDATED, (data) => {
+                emitter.emit(importUuid, data);
+                emitter.emit(importUuid, {
+                    event: 'item-updated',
+                    data,
+                });
+                onItemUpdated(data).catch((err) => errors.push(err));
+            });
         }
 
         bootstrapper.on(EVENT_NAMES.DONE, () => {
+            emitter.emit(importUuid, 'done');
             bootstrapper.kill();
             if (errors.length > 0) {
                 resolve({
@@ -69,8 +91,13 @@ export const runImport = async (
             });
         });
         bootstrapper.on(EVENT_NAMES.ERROR, (err: any) => {
+            emitter.emit(importUuid, {
+                event: 'error',
+                data: err,
+            });
             errors.push(err);
         });
+        emitter.emit(importUuid, 'started');
         bootstrapper.start();
     });
 };

components/crystallize-import/src/core.server/services.server.ts

@@ -0,0 +1,20 @@
+import { EventEmitter } from 'events';
+
+declare global {
+    // eslint-disable-next-line no-var
+    var __services: Awaited<ReturnType<typeof build>>;
+}
+
+export const buildServices = async () => {
+    if (!global.__services) {
+        global.__services = await build();
+    }
+    return global.__services;
+};
+
+const build = async () => {
+    const emitter = new EventEmitter();
+    return {
+        emitter,
+    };
+};

components/crystallize-import/src/routes/_index.tsx

@@ -29,6 +29,7 @@ export const loader = async ({ request }: LoaderFunctionArgs) => {
 export default function Index() {
     const { shapes, folders, flows, channels } = useLoaderData<typeof loader>();
     const initialState: State = {
+        importId: Math.random().toString(36).substring(7),
         shapes,
         folders,
         flows,

components/crystallize-import/src/routes/api.import.stream.$id.ts

@@ -0,0 +1,22 @@
+import type { LoaderFunctionArgs } from '@remix-run/node';
+import { eventStream } from 'remix-utils/sse/server';
+import { buildServices } from '~/core.server/services.server';
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+    const { emitter } = await buildServices();
+
+    return eventStream(request.signal, (send) => {
+        const handler = (data: any) => {
+            send({ event: 'log', data: JSON.stringify(data) });
+        };
+        emitter.addListener(params.id!, handler);
+        send({ event: 'init', data: 'Started' });
+        const interval = setInterval(() => {
+            send({ event: 'ping', data: 'pong' });
+        }, 10000);
+        return function clear() {
+            emitter.removeListener(params.id!, handler);
+            clearInterval(interval);
+        };
+    });
+}

components/crystallize-import/src/routes/api.submit.ts

@@ -2,29 +2,38 @@ import { type ActionFunctionArgs, json } from '@remix-run/node';
 import { FormSubmission } from '~/contracts/form-submission';
 import { getCookieValue } from '~/core.server/auth.server';
 import { runImport } from '~/core.server/import-runner.server';
+import { buildServices } from '~/core.server/services.server';
 import { specFromFormSubmission } from '~/core.server/spec-from-form-submission.server';
 import CrystallizeAPI from '~/core.server/use-cases/crystallize';
 
 export const action = async ({ request }: ActionFunctionArgs) => {
+    const { emitter } = await buildServices();
     const api = await CrystallizeAPI(request);
     const post: FormSubmission = await request.json();
     const [validationRules, shapes] = await Promise.all([api.fetchValidationsSchema(), api.fetchShapes()]);
     try {
+        const importId = post.importId ?? Math.random().toString(36).substring(7);
         const spec = await specFromFormSubmission(post, shapes);
         const results = await runImport(
+            importId,
             spec,
             {
                 onItemUpdated: async (item) => {
-                    const push = (stageId: string) =>
-                        api.pushItemToFlow(
+                    const push = async (stageId: string) => {
+                        emitter.emit(importId, {
+                            event: 'stage-push',
+                            item,
+                            stageId,
+                        });
+                        await api.pushItemToFlow(
                             {
                                 id: item.id,
                                 language: item.language,
                                 version: post.doPublish === true ? 'published' : 'draft',
                             },
                             stageId,
                         );
-
+                    };
                     const validate = validationRules?.[item.shape.identifier]?.validate;
                     if (!validate) {
                         // no validation
@@ -48,6 +57,7 @@ export const action = async ({ request }: ActionFunctionArgs) => {
                 },
             },
             {
+                emitter,
                 tenantIdentifier: api.apiClient.config.tenantIdentifier,
                 skipPublication: !(post.doPublish === true),
                 sessionId: getCookieValue(request, 'connect.sid'),

components/crystallize-import/src/ui/import/App.tsx

@@ -1,17 +1,30 @@
+import { useEventSource } from 'remix-utils/sse/react';
 import { ActionBar } from './components/action-bar/ActionBar';
 import { DataMatchingForm } from './components/DataForm';
 import { FileChooser } from './components/FileChooser';
 import { useImport } from './provider';
+import { useEffect, useState } from 'react';
 
 export const App = () => {
     const { state, dispatch } = useImport();
     const { shapes, folders } = state;
+    const [streamLogs, setStreamLogs] = useState<any[]>([]);
+    const data = useEventSource(`/api/import/stream/${state.importId}`, { event: 'log' });
+
+    useEffect(() => {
+        if (state.done) {
+            setStreamLogs([]);
+        }
+        if (data) {
+            setStreamLogs((prev) => [...prev, data]);
+        }
+    }, [data, state.done]);
+
     return (
         <div style={{ fontFamily: 'system-ui, sans-serif', lineHeight: '1.4', padding: '20px 50px' }}>
             <ActionBar shapes={shapes} folders={folders} />
 
             <div style={{ marginTop: 200 }} />
-
             {!state.rows?.length ? (
                 <div className="file-chooser-section app-section">
                     <FileChooser
@@ -62,26 +75,46 @@ export const App = () => {
                             </div>
                         </div>
                     )}
-
                     {state.loading && (
-                        <div className="feedback-container">
-                            {state.loading && (
-                                <>
-                                    <div className="loader-wrapper" style={{ transform: 'scale(0.5,0.5)' }}>
-                                        <div className="loader"></div>
+                        <div>
+                            <div className="feedback-container">
+                                <div className="loader-wrapper" style={{ transform: 'scale(0.5,0.5)' }}>
+                                    <div className="loader"></div>
+                                </div>
+                                <span className="import-message">Bip bop, doing stuff...</span>
+                            </div>
+                            <div className="feedback-container">
+                                <div className="app-section">
+                                    <div className="grid">
+                                        <div className="stream-logs">
+                                            <h2>Stream logs</h2>
+                                            <ul>
+                                                {streamLogs.map((log, i) => {
+                                                    const decoded = JSON.parse(log);
+                                                    return (
+                                                        <li key={i}>
+                                                            <pre>{JSON.stringify(decoded, undefined, 2)}</pre>
+                                                        </li>
+                                                    );
+                                                })}
+                                            </ul>
+                                        </div>
                                     </div>
-                                    <span className="import-message">Bip bop, doing stuff...</span>
-                                </>
-                            )}
+                                </div>
+                            </div>
                         </div>
                     )}
-
                     {state.errors && state.errors.length > 0 && (
                         <div className="error">
                             <p>Errors: </p>
                             <pre>{JSON.stringify(state.errors, null, 2)}</pre>
                         </div>
                     )}
+                    {state.done && (
+                        <div className="feedback-container">
+                            <h1>Import completed</h1>
+                        </div>
+                    )}
 
                     {state.preflight && (
                         <div className="feedback-container">

components/crystallize-import/src/ui/import/components/action-bar/Submit.tsx

@@ -47,6 +47,7 @@ export const Submit = () => {
                     dispatch.updateLoading(true);
                     try {
                         const post: FormSubmission = {
+                            importId: state.importId,
                             shapeIdentifier: state.selectedShape.identifier,
                             folderPath: state.selectedFolder.tree?.path ?? '/',
                             groupProductsBy: state.groupProductsBy,