processing flow, rss fixing (#215)

elliotBraem · web-flow · commit 6aaa245b531e · 2025-07-09T21:37:44.000-06:00
* processing flow, rss fixing

* update
diff --git a/apps/app/src/components/processing/ProcessingHistory.tsx b/apps/app/src/components/processing/ProcessingHistory.tsx
@@ -60,21 +60,20 @@ export function ProcessingHistory({
     }),
     jobColumnHelper.accessor("status", {
       header: "Status",
-      cell: (info) => (
-        <Badge
-          variant={
-            info.getValue() === "completed"
-              ? "secondary"
-              : info.getValue() === "failed"
-                ? "destructive"
-                : info.getValue() === "processing"
-                  ? "default"
-                  : "outline"
-          }
-        >
-          {info.getValue()}
-        </Badge>
-      ),
+      cell: (info) => {
+        const status = info.getValue();
+        const variant =
+          status === "completed"
+            ? "secondary"
+            : status === "failed"
+              ? "destructive"
+              : status === "processing"
+                ? "default"
+                : status === "completed_with_errors"
+                  ? "warning"
+                  : "outline";
+        return <Badge variant={variant}>{status}</Badge>;
+      },
     }),
     jobColumnHelper.accessor("startedAt", {
       header: "Started",
@@ -175,7 +174,6 @@ export function ProcessingHistory({
         return (
           <div className="flex items-center gap-2">
             <Button
-              variant="ghost"
               size="sm"
               onClick={() => {
                 setSelectedStep(step);
diff --git a/apps/app/src/components/processing/ProcessingStepDetails.tsx b/apps/app/src/components/processing/ProcessingStepDetails.tsx
@@ -1,7 +1,7 @@
 import { useTweakAndReprocessStep } from "@/lib/api";
 import { ProcessingStep } from "@curatedotfun/types";
 import { format } from "date-fns";
-import { Loader2 } from "lucide-react";
+import { ClipboardCopy, Loader2 } from "lucide-react";
 import { useEffect, useState } from "react";
 import { Badge } from "../ui/badge";
 import { Button } from "../ui/button";
@@ -126,12 +126,22 @@ export function ProcessingStepDetails({
               {step.error && <TabsTrigger value="error">Error</TabsTrigger>}
             </TabsList>
             <TabsContent value="input" className="flex-1 flex flex-col gap-4">
-              <Textarea
-                value={editableInput}
-                onChange={(e) => setEditableInput(e.target.value)}
-                className="flex-1 font-mono text-sm"
-                rows={15}
-              />
+              <div className="relative flex-1">
+                <Textarea
+                  value={editableInput}
+                  onChange={(e) => setEditableInput(e.target.value)}
+                  className="flex-1 font-mono text-sm h-full"
+                  rows={15}
+                />
+                <Button
+                  variant="ghost"
+                  size="icon"
+                  className="absolute top-2 right-2"
+                  onClick={() => navigator.clipboard.writeText(editableInput)}
+                >
+                  <ClipboardCopy className="h-4 w-4" />
+                </Button>
+              </div>
               <Button
                 onClick={() => {
                   tweakAndReprocess({
@@ -150,25 +160,57 @@ export function ProcessingStepDetails({
               </Button>
             </TabsContent>
             <TabsContent value="output" className="flex-1 overflow-hidden">
-              <div className="h-[300px] w-full rounded-md border p-4 overflow-auto">
+              <div className="h-[300px] w-full rounded-md border p-4 overflow-auto relative">
                 <pre className="text-sm">
                   {step.output ? formatJson(step.output) : "No output data"}
                 </pre>
+                {step.output && (
+                  <Button
+                    variant="ghost"
+                    size="icon"
+                    className="absolute top-2 right-2"
+                    onClick={() =>
+                      navigator.clipboard.writeText(formatJson(step.output))
+                    }
+                  >
+                    <ClipboardCopy className="h-4 w-4" />
+                  </Button>
+                )}
               </div>
             </TabsContent>
             {step.config && (
               <TabsContent value="config" className="flex-1 overflow-hidden">
-                <div className="h-[300px] w-full rounded-md border p-4 overflow-auto">
+                <div className="h-[300px] w-full rounded-md border p-4 overflow-auto relative">
                   <pre className="text-sm">{formatJson(step.config)}</pre>
+                  <Button
+                    variant="ghost"
+                    size="icon"
+                    className="absolute top-2 right-2"
+                    onClick={() =>
+                      navigator.clipboard.writeText(formatJson(step.config))
+                    }
+                  >
+                    <ClipboardCopy className="h-4 w-4" />
+                  </Button>
                 </div>
               </TabsContent>
             )}
             {step.error && (
               <TabsContent value="error" className="flex-1 overflow-hidden">
-                <div className="h-[300px] w-full rounded-md border p-4 bg-red-50 dark:bg-red-950 overflow-auto">
+                <div className="h-[300px] w-full rounded-md border p-4 bg-red-50 dark:bg-red-950 overflow-auto relative">
                   <pre className="text-sm text-red-600 dark:text-red-400">
                     {formatJson(step.error)}
                   </pre>
+                  <Button
+                    variant="ghost"
+                    size="icon"
+                    className="absolute top-2 right-2"
+                    onClick={() =>
+                      navigator.clipboard.writeText(formatJson(step.error))
+                    }
+                  >
+                    <ClipboardCopy className="h-4 w-4" />
+                  </Button>
                 </div>
               </TabsContent>
             )}
diff --git a/apps/app/src/components/ui/badge.tsx b/apps/app/src/components/ui/badge.tsx
@@ -21,6 +21,8 @@ const badgeVariants = cva(
           "py-[4px] rounded-md border-none text-sm bg-red-100 min-w-[80px]  text-red-600 leading-6 capitalize font-medium",
         pending:
           "py-[4px] rounded-md border-none text-center text-sm bg-yellow-100 min-w-[80px]  text-yellow-600 leading-6 capitalize font-medium",
+        warning:
+          "border-transparent bg-yellow-500 text-neutral-50 shadow hover:bg-yellow-500/80 dark:bg-yellow-900 dark:text-neutral-50 dark:hover:bg-yellow-900/80",
       },
     },
     defaultVariants: {
diff --git a/apps/app/src/routes/_layout/feed/$feedId/_tabs/processing.tsx b/apps/app/src/routes/_layout/feed/$feedId/_tabs/processing.tsx
@@ -141,7 +141,6 @@ function ProcessingTab() {
                       </TableCell>
                       <TableCell className="text-right">
                         <Button
-                          variant="ghost"
                           size="sm"
                           onClick={() => toggleExpand(submission.tweetId)}
                         >
diff --git a/apps/app/src/store/feed-creation-store.ts b/apps/app/src/store/feed-creation-store.ts
@@ -101,6 +101,7 @@ const feedCreationStateCreator: StateCreator<
             plugin: "@curatedotfun/rss",
             config: {
               apiSecret: "{{RSS_SERVICE_API_SECRET}}",
+              feedId: "",
               feedConfig: {
                 title: "",
                 description: "",
@@ -133,8 +134,10 @@ const feedCreationStateCreator: StateCreator<
           };
           state.feedConfig.outputs.stream.distribute.push(rssDistributor);
         }
+        rssDistributor.config.feedId = values.id;
         rssDistributor.config.feedConfig.title = values.name ?? "";
         rssDistributor.config.feedConfig.description = values.description ?? "";
+        rssDistributor.config.feedConfig.siteUrl = `https://app.curate.fun/feed/${values.id}`;
         rssDistributor.config.serviceUrl = `https://rss.curate.fun/${values.id}`;
         const objectTransform = rssDistributor.transform?.find(
           (t) => t.plugin === "@curatedotfun/object-transform",
diff --git a/packages/core-services/src/services/distribution.service.ts b/packages/core-services/src/services/distribution.service.ts
@@ -2,6 +2,7 @@ import { DistributionResult, RichSubmission } from "@curatedotfun/shared-db";
 import type { ActionArgs, DistributorConfig } from "@curatedotfun/types";
 import { PluginError, PluginErrorCode } from "@curatedotfun/utils";
 import type { Logger } from "pino";
+import { ZodError } from "zod";
 import type { ConfigService } from "./config.service";
 import { logPluginError } from "../utils/error";
 import { sanitizeJson } from "../utils/sanitize";
@@ -49,8 +50,16 @@ export class DistributionService implements IBaseService {
         await plugin.distribute(args);
         return { success: true };
       } catch (error) {
+        let errorMessage =
+          error instanceof Error ? error.message : "Distribution failed";
+        if (error instanceof ZodError) {
+          errorMessage = error.errors
+            .map((e) => `  - [${e.path.join(".")}]: ${e.message} (${e.code})`)
+            .join("\n");
+          errorMessage = `Configuration validation failed:\n${errorMessage}`;
+        }
         const pluginError = new PluginError(
-          error instanceof Error ? error.message : "Distribution failed",
+          errorMessage,
           {
             pluginName,
             operation: "distribute",
diff --git a/packages/core-services/src/services/processing.service.ts b/packages/core-services/src/services/processing.service.ts
@@ -81,48 +81,58 @@ export class ProcessingService implements IBaseService {
     );
 
     let currentContent: unknown = content;
+    const executedSteps: SelectProcessingStep[] = [];
 
     try {
       for (let i = 0; i < plan.length; i++) {
         const planStep = plan[i];
         const stepOrder = i + 1;
 
         if (planStep.type === "transformation") {
-          currentContent = await this.executeTransform(
+          const result = await this.executeTransform(
             job.id,
             currentContent,
             planStep,
             stepOrder,
           );
+          executedSteps.push(result.step);
+          currentContent = result.output;
         } else if (planStep.type === "distribution") {
-          await this.executeDistribution(
+          const step = await this.executeDistribution(
             job.id,
             currentContent,
             planStep,
             stepOrder,
           );
+          executedSteps.push(step);
         }
       }
 
-      await this.processingRepository.updateJob(job.id, {
-        status: "completed",
+      const hasFailures = executedSteps.some((s) => s.status === "failed");
+      const finalStatus = hasFailures ? "completed_with_errors" : "completed";
+
+      const updatedJob = await this.processingRepository.updateJob(job.id, {
+        status: finalStatus,
         completedAt: new Date(),
       });
-      this.logger.info({ jobId: job.id }, "Processing job completed.");
+      this.logger.info(
+        { jobId: job.id, finalStatus },
+        "Processing job finished.",
+      );
+      return updatedJob;
     } catch (error) {
       await this.handleProcessingError(job.id, error);
       throw error;
     }
-    return job;
   }
 
   private async executeTransform(
     jobId: string,
     input: unknown,
     planStep: ProcessingPlanStep,
     stepOrder: number,
-  ): Promise<unknown> {
-    const step = await this.processingRepository.createStep({
+  ): Promise<{ output: unknown; step: SelectProcessingStep }> {
+    let step = await this.processingRepository.createStep({
       jobId,
       stepOrder,
       type: "transformation",
@@ -139,12 +149,12 @@ export class ProcessingService implements IBaseService {
       const output = await this.transformationService.applyTransforms(input, [
         planStep as TransformConfig,
       ]);
-      await this.processingRepository.updateStep(step.id, {
+      const updatedStep = await this.processingRepository.updateStep(step.id, {
         status: "success",
         output: output as Json,
         completedAt: new Date(),
       });
-      return output;
+      return { output, step: updatedStep };
     } catch (error) {
       const stepError: StepError = {
         message: error instanceof Error ? error.message : String(error),
@@ -172,7 +182,7 @@ export class ProcessingService implements IBaseService {
     input: unknown,
     planStep: ProcessingPlanStep,
     stepOrder: number,
-  ) {
+  ): Promise<SelectProcessingStep> {
     const step = await this.processingRepository.createStep({
       jobId,
       stepOrder,
@@ -191,7 +201,7 @@ export class ProcessingService implements IBaseService {
         planStep as DistributorConfig,
         input,
       );
-      await this.processingRepository.updateStep(step.id, {
+      return this.processingRepository.updateStep(step.id, {
         status: "success",
         output: output as Json,
         completedAt: new Date(),
@@ -202,12 +212,15 @@ export class ProcessingService implements IBaseService {
         stack: error instanceof Error ? error.stack : undefined,
         stepName: planStep.plugin,
       };
-      await this.processingRepository.updateStep(step.id, {
+      this.logger.warn(
+        { jobId, stepId: step.id, plugin: planStep.plugin, error },
+        "Distribution step failed. Continuing processing.",
+      );
+      return this.processingRepository.updateStep(step.id, {
         status: "failed",
         error: stepError as Json,
         completedAt: new Date(),
       });
-      throw error;
     }
   }
 
diff --git a/packages/shared-db/migrations/0022_supreme_vertigo.sql b/packages/shared-db/migrations/0022_supreme_vertigo.sql
@@ -0,0 +1,3 @@
+ALTER TYPE "public"."processing_job_status" ADD VALUE 'completed_with_errors';
+DROP INDEX "public"."processing_jobs_retry_check";
+ALTER TABLE "public"."processing_jobs" ADD CONSTRAINT "processing_jobs_retry_check" CHECK (id != retry_of_job_id OR retry_of_job_id IS NULL);
diff --git a/packages/shared-db/migrations/meta/0022_snapshot.json b/packages/shared-db/migrations/meta/0022_snapshot.json
diff --git a/packages/shared-db/migrations/meta/_journal.json b/packages/shared-db/migrations/meta/_journal.json
diff --git a/packages/shared-db/src/schema/processing.ts b/packages/shared-db/src/schema/processing.ts
diff --git a/packages/types/src/api/processing.ts b/packages/types/src/api/processing.ts

Original file line number	Diff line number	Diff line change
`@@ -141,7 +141,6 @@ function ProcessingTab() {`
`141`	`141`	`</TableCell>`
`142`	`142`	`<TableCell className="text-right">`
`143`	`143`	`<Button`
`144`		`- variant="ghost"`
`145`	`144`	`size="sm"`
`146`	`145`	`onClick={() => toggleExpand(submission.tweetId)}`
`147`	`146`	`>`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+ALTER TYPE "public"."processing_job_status" ADD VALUE 'completed_with_errors';`
	`2`	`+DROP INDEX "public"."processing_jobs_retry_check";`
	`3`	`+ALTER TABLE "public"."processing_jobs" ADD CONSTRAINT "processing_jobs_retry_check" CHECK (id != retry_of_job_id OR retry_of_job_id IS NULL);`