feat: enable requesting recommendations

packages/client/src/components/ProfilePage.tsx

@@ -8,6 +8,7 @@ import {
   Button,
   Collapse,
   Paper,
+  Snackbar,
   Table,
   TableBody,
   TableCell,
@@ -27,7 +28,8 @@ interface ProfilePageProps {
   auth: AuthInfo | undefined;
 }
 
-interface TableRow {
+interface SummaryRowData {
+  id: number;
   createdAt: string;
   type: string;
   content: string;
@@ -48,6 +50,8 @@ export function ProfilePage(props: ProfilePageProps) {
     React.useState<RouterOutput["getPublicUser"]>();
 
   const [isFollowing, setIsFollowing] = React.useState<boolean>(false);
+  const [requestingRecommendations, setRequestingRecommendations] =
+    React.useState(false);
 
   React.useEffect(() => {
     if (!props.auth?.authenticated) return;
@@ -98,28 +102,37 @@ export function ProfilePage(props: ProfilePageProps) {
     );
   }
 
-  const rows: TableRow[] = [];
+  const rows: SummaryRowData[] = [];
   summaries?.forEach((summary) => {
     rows.push({
+      id: summary.id,
       createdAt: summary.createdAt,
       type: "Twitter Summary",
       content: summary.content,
       useForRecommendations: summary.useForRecommendations,
     });
   });
 
+  const [customQuery, setCustomQuery] = React.useState<string>("");
+  const [customQueryMessage, setCustomQueryMessage] =
+    React.useState<string>("");
+  const handleClose = () => {
+    setCustomQueryMessage("");
+  };
+
   if (!profileForUser) {
     return <div>Loading...</div>;
   }
 
-  profileForUser.following.forEach((follow) => {
-    rows.push({
-      createdAt: follow.createdAt,
-      type: "Following",
-      content: follow.user.username,
-      useForRecommendations: true,
-    });
-  });
+  // profileForUser.following.forEach((follow) => {
+  //   rows.push({
+  //     id: follow.id,
+  //     createdAt: follow.createdAt,
+  //     type: "Following",
+  //     content: follow.user.username,
+  //     useForRecommendations: true,
+  //   });
+  // });
 
   return (
     <div className="p-4">
@@ -153,14 +166,43 @@ export function ProfilePage(props: ProfilePageProps) {
       </div>
       <br></br>
 
-      <div className="flex items-center gap-2">
-        <TextField
-          id="outlined-basic"
-          label="Custom Query"
-          variant="outlined"
-        />
-        <Button variant="contained">Search</Button>
-      </div>
+      {viewingOwnProfile && (
+        <div className="flex items-center gap-2">
+          <TextField
+            disabled={requestingRecommendations}
+            id="outlined-basic"
+            label="Custom Query"
+            variant="outlined"
+            value={customQuery}
+            onChange={(e) => setCustomQuery(e.target.value)}
+          />
+          <Button
+            disabled={!customQuery || requestingRecommendations}
+            onClick={async () => {
+              setRequestingRecommendations(true);
+              const res = await trpc.requestRecommendations.mutate({
+                customQuery,
+              });
+              console.log(res);
+              if (res.type === "success") {
+                setCustomQueryMessage("Recommendations pipeline started");
+              } else {
+                setCustomQueryMessage(res.error);
+              }
+              setRequestingRecommendations(false);
+            }}
+            variant="contained"
+          >
+            Search
+          </Button>
+          <Snackbar
+            open={!!customQueryMessage}
+            autoHideDuration={6000}
+            onClose={handleClose}
+            message={customQueryMessage}
+          />
+        </div>
+      )}
       <br></br>
       <h3>Recommendation Inputs</h3>
       <br></br>
@@ -177,9 +219,12 @@ export function ProfilePage(props: ProfilePageProps) {
           <TableBody>
             {_.sortBy(rows, (x) => x.createdAt).map((row, idx) => (
               <SummaryRow
+                requestingRecommendations={requestingRecommendations}
+                setRequestingRecommendations={setRequestingRecommendations}
                 key={idx}
                 row={row}
                 viewingOwnProfile={!!viewingOwnProfile}
+                auth={props.auth}
               />
             ))}
           </TableBody>
@@ -190,12 +235,15 @@ export function ProfilePage(props: ProfilePageProps) {
 }
 
 interface SummaryRowProps {
-  row: TableRow;
+  row: SummaryRowData;
   viewingOwnProfile: boolean;
+  auth: AuthInfo | undefined;
+  requestingRecommendations: boolean;
+  setRequestingRecommendations: (value: boolean) => void;
 }
 
 function SummaryRow(props: SummaryRowProps) {
-  const { row, viewingOwnProfile } = props;
+  const { row, viewingOwnProfile, auth } = props;
   const [expanded, setExpanded] = React.useState(false);
   return (
     <TableRow>
@@ -218,16 +266,60 @@ function SummaryRow(props: SummaryRowProps) {
           </div>
         </Collapse>
       </TableCell>
-      {viewingOwnProfile && (
+      {viewingOwnProfile && auth?.authenticated && (
         <TableCell>
           <div>{/* <Button>Edit</Button> */}</div>
           <div>
-            <Button onClick={() => {}} variant="contained">
-              Get Recommendations
-            </Button>
+            <GetRecommendationsButton
+              setRequestingRecommendations={props.setRequestingRecommendations}
+              row={row}
+              auth={auth}
+            />
           </div>
         </TableCell>
       )}
     </TableRow>
   );
 }
+
+interface GetRecommendationsButtonProps {
+  row: SummaryRowData;
+  auth: AuthInfo | undefined;
+  disabled?: boolean;
+  setRequestingRecommendations: (value: boolean) => void;
+}
+
+function GetRecommendationsButton(props: GetRecommendationsButtonProps) {
+  const [message, setMessage] = React.useState("");
+  const handleClose = () => {
+    setMessage("");
+  };
+  return (
+    <>
+      <Button
+        disabled={props.disabled}
+        onClick={async () => {
+          props.setRequestingRecommendations(true);
+          const res = await trpc.requestRecommendations.mutate({
+            summaryId: props.row.id,
+          });
+          if (res.type === "success") {
+            setMessage("Recommendations pipeline started");
+          } else {
+            setMessage(res.error);
+          }
+          props.setRequestingRecommendations(false);
+        }}
+        variant="contained"
+      >
+        Get Recommendations
+      </Button>
+      <Snackbar
+        open={!!message}
+        autoHideDuration={6000}
+        onClose={handleClose}
+        message={message}
+      />
+    </>
+  );
+}

packages/server/src/lib/getNumRunningPipelines.ts

@@ -0,0 +1,26 @@
+import { UserModel } from "shared/src/schemas";
+import { z } from "zod";
+import { prisma } from "../db";
+
+export async function getNumRunningPipelines(
+  authenticatedUser: z.infer<typeof UserModel>
+): Promise<number | undefined> {
+  if (!authenticatedUser) {
+    return;
+  }
+  const pipelines = (
+    await prisma.pipelineRun.findMany({
+      where: {
+        username: authenticatedUser.username,
+      },
+      include: {
+        tasks: true,
+      },
+    })
+  )
+    // TODO: check
+    // slice(1) because the first task is the pipeline itself, doesn't get updated properly
+    .filter((p) => p.tasks.slice(1).some((t) => t.status === "running"));
+
+  return pipelines.length;
+}

packages/server/src/routers/authenticatedRouter.ts

@@ -3,6 +3,9 @@ import { router, publicProcedure } from "../trpc";
 import { z } from "zod";
 import { PublicUserModel } from "shared/src/manual/PublicUser";
 import { generateAPIKey } from "../generateAPIKey";
+import { addPipeline } from "../tasks/worker";
+import { randomUUID } from "crypto";
+import { getNumRunningPipelines } from "../lib/getNumRunningPipelines";
 
 export const authenticatedRouter = router({
   voteOnRecommendation: publicProcedure
@@ -326,4 +329,61 @@ export const authenticatedRouter = router({
     });
     return apiKey;
   }),
+
+  getNumRunningPipelines: publicProcedure.query(async ({ ctx }) => {
+    const user = await prisma.user.findUnique({
+      where: {
+        id: ctx.user?.id,
+      },
+    });
+    if (!user) {
+      return;
+    }
+    return await getNumRunningPipelines(user);
+  }),
+
+  requestRecommendations: publicProcedure
+    .input(
+      z.object({
+        summaryId: z.number().optional(),
+        customQuery: z.string().optional(),
+      })
+    )
+    .mutation(async ({ ctx, input }) => {
+      const authenticatedUser = await prisma.user.findUnique({
+        where: {
+          id: ctx.user?.id,
+        },
+      });
+      if (!authenticatedUser) {
+        return { type: "error" as const, error: "User not found" };
+      }
+
+      const numRunning = await getNumRunningPipelines(authenticatedUser);
+      if (numRunning && numRunning >= 2) {
+        return {
+          type: "error" as const,
+          error: "You can only have max 2 pipelines running at a time",
+        };
+      }
+
+      const summary = input.summaryId
+        ? await prisma.summary.findFirst({
+            where: {
+              userId: authenticatedUser.id,
+              id: input.summaryId,
+            },
+          })
+        : undefined;
+
+      const job = await addPipeline("twitter-pipeline-v1", {
+        username: authenticatedUser.username,
+        summary: summary?.content,
+        queries: input.customQuery ? [input.customQuery] : undefined,
+        runId: randomUUID(),
+        emailResults: true,
+      });
+
+      return { type: "success" as const };
+    }),
 });

packages/server/src/tasks/twitterPipeline.saga.ts

@@ -35,7 +35,6 @@ import { sendEmail } from "../lib/sendEmail";
 import { TranscriptClipWithScore } from "shared/src/manual/TranscriptClip";
 import { ArticleSnippetWithScore } from "shared/src/manual/ArticleSnippet";
 import { chunksToClips } from "cli/src/recommender/chunksToClips";
-import { format } from "path";
 
 type QueryWithSearchResultWithTranscript = {
   searchResults: (VideoResultWithTranscript | MetaphorArticleResult)[];