set the port of pipes (#1128)

* update pipe port

* fix

screenpipe-app-tauri/components/pipe-config-form.tsx

@@ -62,62 +62,6 @@ export const PipeConfigForm: React.FC<PipeConfigFormProps> = ({
     }));
   };
 
-  const handleSubmit = async (e: React.FormEvent) => {
-    e.preventDefault();
-    console.log("submitting config:", config);
-
-    if (!config?.fields) {
-      console.log("no config fields found, aborting");
-      return;
-    }
-
-    try {
-      toast({
-        title: "updating pipe configuration",
-        description: "please wait...",
-      });
-
-      if (!pipe.id) {
-        throw new Error("pipe id is missing");
-      }
-
-      const response = await fetch(`http://localhost:3030/pipes/update`, {
-        method: "POST",
-        headers: {
-          "Content-Type": "application/json",
-        },
-        body: JSON.stringify({
-          pipe_id: pipe.id,
-          config: config,
-        }),
-      });
-
-      if (!response.ok) {
-        const errorText = await response.text();
-        throw new Error(`failed to update pipe config: ${errorText}`);
-      }
-
-      const result = await response.json();
-      console.log("update response:", result);
-
-      onConfigSave(config);
-
-      await new Promise((resolve) => setTimeout(resolve, 1500));
-
-      toast({
-        title: "Configuration updated",
-        description: "The pipe configuration has been successfully updated.",
-      });
-    } catch (error) {
-      console.error("Error saving pipe config:", error);
-      toast({
-        title: "Error updating configuration",
-        description: "Failed to update pipe configuration. Please try again.",
-        variant: "destructive",
-      });
-    }
-  };
-
   const renderConfigInput = (field: FieldConfig) => {
     const value = field?.value ?? field?.default;
 
@@ -407,13 +351,98 @@ export const PipeConfigForm: React.FC<PipeConfigFormProps> = ({
   };
 
   return (
-    <form onSubmit={handleSubmit} className="space-y-6">
+    <div className="space-y-6">
       <h3 className="text-lg font-semibold">pipe configuration</h3>
-      {config?.fields && config.fields.length > 0 && (
-        <Button type="submit" onClick={handleSubmit}>
-          save configuration
-        </Button>
+
+      {config?.is_nextjs && (
+        <div className="space-y-2">
+          <Label htmlFor="port" className="font-medium">
+            port (number)
+          </Label>
+          <div className="flex items-center space-x-2">
+            <Input
+              id="port"
+              type="number"
+              value={config.port ?? ''}
+              onChange={(e) => setConfig(prev => prev ? {
+                ...prev,
+                port: parseInt(e.target.value) || 3000
+              } : prev)}
+              onWheel={(e) => e.preventDefault()}
+              step="1"
+              min="1"
+              max="65535"
+              autoCorrect="off"
+              spellCheck="false"
+            />
+            <TooltipProvider>
+              <Tooltip>
+                <TooltipTrigger asChild>
+                  <Button
+                    size="icon"
+                    variant="ghost"
+                    onClick={() => setConfig(prev => prev ? {...prev, port: 3000} : prev)}
+                    className="h-8 w-8"
+                  >
+                    <RefreshCw className="h-4 w-4" />
+                  </Button>
+                </TooltipTrigger>
+                <TooltipContent>
+                  <p>Reset to default (3000)</p>
+                </TooltipContent>
+              </Tooltip>
+            </TooltipProvider>
+          </div>
+          <MemoizedReactMarkdown
+            className="prose prose-sm break-words dark:prose-invert prose-p:leading-relaxed prose-pre:p-0 w-full"
+            remarkPlugins={[remarkGfm, remarkMath]}
+            components={{
+              p({ children }) {
+                return <p className="mb-2 last:mb-0">{children}</p>;
+              },
+              a({ node, href, children, ...props }) {
+                return (
+                  <a
+                    href={href}
+                    target="_blank"
+                    rel="noopener noreferrer"
+                    {...props}
+                  >
+                    {children}
+                  </a>
+                );
+              },
+              code({ node, className, children, ...props }) {
+                const content = String(children).replace(/\n$/, "");
+                const match = /language-(\w+)/.exec(className || "");
+
+                if (!match) {
+                  return (
+                    <code
+                      className="px-1 py-0.5 rounded-sm font-mono text-sm"
+                      {...props}
+                    >
+                      {content}
+                    </code>
+                  );
+                }
+
+                return (
+                  <CodeBlock
+                    key={Math.random()}
+                    language={(match && match[1]) || ""}
+                    value={content}
+                    {...props}
+                  />
+                );
+              },
+            }}
+          >
+            Port number for this pipe. If the selected port is already in use when starting the pipe, a random available port will be automatically assigned.
+          </MemoizedReactMarkdown>
+        </div>
       )}
+
       {config?.fields?.map((field: FieldConfig) => (
         <div key={field.name} className="space-y-2">
           <Label htmlFor={field.name} className="font-medium">
@@ -469,6 +498,9 @@ export const PipeConfigForm: React.FC<PipeConfigFormProps> = ({
           </MemoizedReactMarkdown>
         </div>
       ))}
-    </form>
+        <Button type="submit" onClick={() => onConfigSave(config || {})}>
+          save configuration
+        </Button>
+    </div>
   );
 };

screenpipe-app-tauri/components/pipe-store.tsx

@@ -523,6 +523,19 @@ const PipeStore: React.FC = () => {
     }
   };
 
+  const reloadPipeConfig = async (pipe: Pipe) => {
+    await fetchInstalledPipes();
+
+    const freshPipe = pipes.find(
+      (p) => normalizeId(p.id) === normalizeId(pipe.id)
+    );
+    if (freshPipe) {
+      console.log("freshPipe", freshPipe);
+
+      setSelectedPipe(freshPipe);
+    }
+  };
+
   const handleToggleEnabled = async (pipe: Pipe) => {
     try {
       // Reset broken state when manually toggling
@@ -560,7 +573,8 @@ const PipeStore: React.FC = () => {
           return;
         }
 
-        const hasSubscription = await checkExistingSubscription(pipe.id);
+        // const hasSubscription = await checkExistingSubscription(pipe.id);
+        const hasSubscription = true;
         console.log("subscription check:", {
           hasSubscription,
           pipeId: pipe.id,
@@ -788,6 +802,8 @@ const PipeStore: React.FC = () => {
           title: "Configuration saved",
           description: "The pipe configuration has been updated.",
         });
+
+        await setSelectedPipe({...selectedPipe, config: config});
       } catch (error) {
         console.error("Failed to save config:", error);
         toast({
@@ -1198,7 +1214,7 @@ const PipeStore: React.FC = () => {
               </div>
 
               {selectedPipe.enabled &&
-                selectedPipe.config?.fields?.length > 0 && (
+                 (
                   <div className="space-y-3 pt-4 border-t">
                     <PipeConfigForm
                       pipe={selectedPipe}

screenpipe-core/src/pipes.rs

@@ -131,15 +131,32 @@ mod pipes {
                 "setting up next.js specific configuration for pipe: {}",
                 pipe
             );
+
+            let mut assigned_port = None;
+
             // Handle Next.js specific setup including crons
             if pipe_json_path.exists() {
                 debug!("reading pipe.json for next.js configuration");
                 let pipe_json = tokio::fs::read_to_string(&pipe_json_path).await?;
                 let pipe_config: Value = serde_json::from_str(&pipe_json)?;
 
-                // Update pipe.json with the port
-                let port = pick_unused_port().expect("No ports free");
-                debug!("picked unused port {} for next.js pipe", port);
+                // Try to use user-configured port first
+                if let Some(user_port) = pipe_config.get("port").and_then(|p| p.as_u64()) {
+                    debug!("found user-configured port: {}", user_port);
+                    // Verify port is available
+                    if is_port_available(user_port as u16) {
+                        assigned_port = Some(user_port as u16);
+                        debug!("user-configured port {} is available", user_port);
+                    } else {
+                        debug!("user-configured port {} is in use, will assign random port", user_port);
+                    }
+                }
+
+                // Fallback to random port if needed
+                let port = assigned_port.unwrap_or_else(|| pick_unused_port().expect("No ports free"));
+                info!("using port {} for next.js pipe", port);
+
+                // Update pipe.json with the actual port being used
                 let mut updated_config = pipe_config.clone();
                 updated_config["port"] = json!(port);
                 let updated_pipe_json = serde_json::to_string_pretty(&updated_config)?;
@@ -216,7 +233,7 @@ mod pipes {
                 debug!("successfully installed dependencies for next.js pipe");
             } else {
                 let port = pick_unused_port().expect("No ports free");
-                debug!("no pipe.json found, using port {} for next.js pipe", port);
+                debug!("no pipe.json found, using random port {} for next.js pipe", port);
                 env_vars.push(("PORT".to_string(), port.to_string()));
             }
 
@@ -1032,6 +1049,12 @@ mod pipes {
             Ok(false)
         }
     }
+
+    // Add this helper function to check if a port is available
+    fn is_port_available(port: u16) -> bool {
+        use std::net::TcpListener;
+        TcpListener::bind(("127.0.0.1", port)).is_ok()
+    }
 }
 
 #[cfg(feature = "pipes")]

screenpipe-server/src/pipe_manager.rs

@@ -77,10 +77,19 @@ impl PipeManager {
 
         debug!("is_enabled: {}", is_enabled.unwrap_or(false));
 
+        // Handle both top-level properties and nested fields
         if let Value::Object(existing_config) = &mut config {
             if let Value::Object(updates) = new_config {
-                for (key, value) in updates {
-                    existing_config.insert(key, value);
+                // Update top-level properties
+                for (key, value) in updates.iter() {
+                    if key != "fields" {  // Handle non-fields properties directly
+                        existing_config.insert(key.clone(), value.clone());
+                    }
+                }
+
+                // Handle fields separately if they exist
+                if let Some(Value::Array(new_fields)) = updates.get("fields") {
+                    existing_config.insert("fields".to_string(), Value::Array(new_fields.clone()));
                 }
             } else {
                 return Err(anyhow::anyhow!("new configuration must be an object"));