Enhance demo functionality and improve error handling

- Updated the `justfile` to include a new worker command for processing uploaded documents, improving local development setup.
- Adjusted logging behavior in the API to throttle log messages based on processing counts, enhancing performance monitoring.
- Modified API handlers to support Azure endpoint configuration, improving flexibility in AI integrations.
- Enhanced error handling in the document processing pipeline and UI components to provide clearer feedback to users during uploads and queries.
- Improved the user experience by adding contextual messages for empty responses and refining error messages in the frontend.

These changes aim to streamline the development process and enhance the overall user experience in document management and AI configuration.

Author: rithulkamesh

demo/go/internal/api/documents.go

@@ -151,10 +151,10 @@ func (h *Handler) listDocuments(w http.ResponseWriter, r *http.Request) {
 	}
 	if processingCount > 0 {
 		now := time.Now().UnixNano()
-		throttleNanos := int64(30 * time.Second)
+		throttleNanos := int64(10 * time.Second)
 		listLogMu.Lock()
 		state, ok := listLogState[projectID]
-		shouldLog := !ok || state.lastCount != processingCount || (now-state.lastLogUnix) > throttleNanos
+		shouldLog := !ok || (now-state.lastLogUnix) > throttleNanos
 		if shouldLog {
 			listLogState[projectID] = struct{ lastLogUnix int64; lastCount int }{lastLogUnix: now, lastCount: processingCount}
 		}

demo/go/internal/api/handler.go

@@ -30,7 +30,7 @@ func NewHandler(cfg *config.Config, pool *db.Pool, store *blob.Store, pub *mq.Pu
 func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// CORS for local/dev; tighten in production
 	w.Header().Set("Access-Control-Allow-Origin", "*")
-	w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PATCH, DELETE, OPTIONS")
+	w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS")
 	w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
 	if r.Method == http.MethodOptions {
 		w.WriteHeader(http.StatusNoContent)
@@ -147,7 +147,8 @@ func (h *Handler) putAIConfig(w http.ResponseWriter, r *http.Request) {
 }
 
 // openaiClientFromDBConfig builds an OpenAI client from DB config (for query/stream when no key in body).
-func openaiClientFromDBConfig(cfg *db.AIConfigDecrypted) (*openai.Client, string) {
+// envAzureEndpoint is used when provider is Azure and DB endpoint is empty (e.g. key in Settings, endpoint in .env).
+func openaiClientFromDBConfig(cfg *db.AIConfigDecrypted, envAzureEndpoint string) (*openai.Client, string) {
 	if cfg == nil || cfg.APIKey == "" {
 		return nil, ""
 	}
@@ -157,7 +158,10 @@ func openaiClientFromDBConfig(cfg *db.AIConfigDecrypted) (*openai.Client, string
 	}
 	switch cfg.Provider {
 	case "azure":
-		endpoint := cfg.Endpoint
+		endpoint := strings.TrimSpace(cfg.Endpoint)
+		if endpoint == "" {
+			endpoint = strings.TrimSpace(envAzureEndpoint)
+		}
 		if endpoint == "" {
 			return nil, ""
 		}
@@ -204,7 +208,7 @@ func (h *Handler) query(w http.ResponseWriter, r *http.Request) {
 	} else if len(h.cfg.EncryptionKey) == 32 {
 		dbCfg, _ := h.pool.GetAIConfigDecrypted(r.Context(), h.cfg.EncryptionKey)
 		if dbCfg != nil && dbCfg.APIKey != "" {
-			chatClient, model = openaiClientFromDBConfig(dbCfg)
+			chatClient, model = openaiClientFromDBConfig(dbCfg, h.cfg.AzureEndpoint)
 			if model == "" {
 				model = dbCfg.Model
 			}
@@ -289,13 +293,17 @@ func (h *Handler) queryStream(w http.ResponseWriter, r *http.Request) {
 	} else if len(h.cfg.EncryptionKey) == 32 {
 		dbCfg, _ := h.pool.GetAIConfigDecrypted(ctx, h.cfg.EncryptionKey)
 		if dbCfg != nil && dbCfg.APIKey != "" {
-			streamClient, model = openaiClientFromDBConfig(dbCfg)
+			streamClient, model = openaiClientFromDBConfig(dbCfg, h.cfg.AzureEndpoint)
 			if model == "" {
 				model = dbCfg.Model
 			}
 		}
 	}
 	if err := h.rag.StreamCompletionWithClient(ctx, prompt, w, streamClient, model); err != nil {
+		_ = enc.Encode(map[string]any{"error": err.Error()})
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
 		return
 	}
 }

demo/go/internal/config/config.go

@@ -1,12 +1,17 @@
 package config
 
 import (
+	"log"
 	"os"
+	"strings"
 
 	"github.com/rithulkamesh/docproc/demo/internal/secure"
 	"github.com/sashabaranov/go-openai"
 )
 
+// DefaultEncryptionPassphrase is used when DOCPROC_ENCRYPTION_KEY is not set (dev only; set your own in production).
+const DefaultEncryptionPassphrase = "docproc-dev-default-change-in-production"
+
 // Config holds demo app configuration (env or file).
 type Config struct {
 	DatabaseURL string // PostgreSQL (documents, projects, pgvector)
@@ -36,7 +41,7 @@ func Load() (*Config, error) {
 		S3Region:    getEnv("AWS_REGION", "us-east-1"),
 		MQURL:       getEnv("MQ_URL", "amqp://docproc:docproc@localhost:5672/"),
 		DocprocCLI:  getEnv("DOCPROC_CLI", "docproc"),
-		EncryptionKey: keyFromEnvOrNil(os.Getenv("DOCPROC_ENCRYPTION_KEY")),
+		EncryptionKey: keyFromEnvOrDefault(os.Getenv("DOCPROC_ENCRYPTION_KEY")),
 		OpenAIKey:   os.Getenv("OPENAI_API_KEY"),
 		OpenAIModel: getEnv("OPENAI_MODEL", "gpt-4o-mini"),
 		AzureAPIKey:             os.Getenv("AZURE_OPENAI_API_KEY"),
@@ -103,10 +108,11 @@ func getEnv(key, defaultVal string) string {
 	return defaultVal
 }
 
-// keyFromEnvOrNil returns a 32-byte key derived from passphrase, or nil if passphrase is empty.
-func keyFromEnvOrNil(passphrase string) []byte {
-	if passphrase == "" {
-		return nil
+// keyFromEnvOrDefault returns a 32-byte key derived from passphrase. If passphrase is empty, uses a default (dev only; log warning).
+func keyFromEnvOrDefault(passphrase string) []byte {
+	if strings.TrimSpace(passphrase) == "" {
+		log.Print("[config] DOCPROC_ENCRYPTION_KEY not set; using default (fine for dev; set your own in production)")
+		return secure.KeyFromEnv(DefaultEncryptionPassphrase)
 	}
 	return secure.KeyFromEnv(passphrase)
 }

demo/go/internal/db/db.go

@@ -2,13 +2,24 @@ package db
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 	pgxvec "github.com/pgvector/pgvector-go/pgx"
 )
 
+// isDuplicateExtensionError returns true if err is a unique violation on pg_extension (extension already exists).
+func isDuplicateExtensionError(err error) bool {
+	var pgErr *pgconn.PgError
+	if !errors.As(err, &pgErr) {
+		return false
+	}
+	return pgErr.Code == "23505" && pgErr.ConstraintName == "pg_extension_name_index"
+}
+
 // Pool is the PostgreSQL connection pool.
 type Pool struct {
 	*pgxpool.Pool
@@ -28,7 +39,10 @@ func NewPool(ctx context.Context, connString string) (*Pool, error) {
 	_, err = conn.Exec(ctx, `CREATE EXTENSION IF NOT EXISTS vector`)
 	conn.Close(ctx)
 	if err != nil {
-		return nil, fmt.Errorf("vector extension: %w", err)
+		// Another process (e.g. API and worker both starting) may have created it; duplicate key is OK.
+		if !isDuplicateExtensionError(err) {
+			return nil, fmt.Errorf("vector extension: %w", err)
+		}
 	}
 	config.AfterConnect = func(ctx context.Context, c *pgx.Conn) error {
 		return pgxvec.RegisterTypes(ctx, c)

demo/go/internal/rag/rag.go

@@ -99,7 +99,7 @@ func (r *RAG) QueryWithClient(ctx context.Context, question string, client *open
 		return "", sources, fmt.Errorf("chat: %w", err)
 	}
 	if len(chatResp.Choices) == 0 {
-		return "", sources, nil
+		return "", sources, fmt.Errorf("model returned no choices")
 	}
 	answer = strings.TrimSpace(chatResp.Choices[0].Message.Content)
 	return answer, sources, nil

demo/justfile

@@ -10,9 +10,10 @@ deps:
     docker compose up -d
     @echo "Waiting for postgres..."
     sleep 3
-    @echo "Deps up. Run: just dev  (API + web in one terminal, Ctrl+C kills both)"
+    @echo "Deps up. Run: just dev  (API + worker + web in one terminal, Ctrl+C kills all)"
 
-# Run API + web locally; logs from both in one terminal. Ctrl+C kills both.
+# Run API + web + worker locally; logs from all in one terminal. Ctrl+C kills all.
+# Worker processes uploaded documents (needs docproc CLI on PATH: pip install docproc, or python -m docproc.bin.cli).
 dev: deps
     #!/usr/bin/env bash
     set -e
@@ -21,6 +22,7 @@ dev: deps
     export S3_ENDPOINT="http://localhost:4566"
     trap 'kill $(jobs -p) 2>/dev/null; exit 130' INT TERM
     (cd go && go run .) &
+    (cd go && go run . -worker) &
     (cd web && VITE_DOCPROC_API_URL=http://localhost:8080 npm run dev) &
     wait
 
@@ -35,6 +37,14 @@ api:
     export S3_ENDPOINT="http://localhost:4566"
     cd go && go run .
 
+# Run document processing worker only. Consumes jobs from RabbitMQ, runs docproc CLI.
+# Requires docproc on PATH (e.g. pip install -e . from repo root, or set DOCPROC_CLI).
+worker:
+    export DATABASE_URL="postgresql://docproc:docproc@localhost:5432/docproc?sslmode=disable"
+    export MQ_URL="amqp://docproc:docproc@localhost:5672/"
+    export S3_ENDPOINT="http://localhost:4566"
+    cd go && go run . -worker
+
 # Run Go API with live reload (install: go install github.com/air-verse/air@latest)
 api-watch:
     export DATABASE_URL="postgresql://docproc:docproc@localhost:5432/docproc?sslmode=disable"

demo/web/src/api/query.ts

@@ -78,6 +78,10 @@ export async function runQueryStream(
         if (!trimmed) continue
         try {
           const data = JSON.parse(trimmed) as Record<string, unknown>
+          if (typeof data.error === 'string') {
+            callbacks.onError(data.error)
+            return false
+          }
           if (Array.isArray(data.sources)) {
             callbacks.onSources(data.sources as RagSource[])
           } else if (typeof data.delta === 'string') {

demo/web/src/components/canvas/ConverseCanvas.tsx

@@ -93,7 +93,7 @@ function normalizeQueryError(msg: string): string {
     return "AI provider isn't set up correctly. Check your config."
   }
   if (msg.includes('api_key') || msg.includes('401') || msg.includes('403')) {
-    return 'API key missing or invalid.'
+    return 'API key missing or invalid. Set OPENAI_API_KEY or AZURE_OPENAI_API_KEY in .env, or add your key in Settings → AI provider.'
   }
   return msg
 }
@@ -211,6 +211,13 @@ export function ConverseCanvas() {
         },
         onDone: () => {
           stopStreamFlush()
+          setMessages((prev) =>
+            prev.map((m) =>
+              m.id === assistantId && !m.content.trim()
+                ? { ...m, content: 'No response from the model. Check your AI config or try again.' }
+                : m
+            )
+          )
           setSending(false)
         },
         onError: applyError,
@@ -221,6 +228,15 @@ export function ConverseCanvas() {
         if (res.answer.startsWith('Query failed:')) {
           const raw = res.answer.replace(/^Query failed:\s*/, '').trim()
           applyError(raw)
+        } else if (!res.answer.trim()) {
+          setMessages((prev) =>
+            prev.map((m) =>
+              m.id === assistantId
+                ? { ...m, content: 'No response from the model. Check your AI config or try again.' }
+                : m
+            )
+          )
+          setSending(false)
         } else {
           // Simulate streaming for non-streaming API so UX is consistent
           const full = res.answer

demo/web/src/components/canvas/SourcesCanvas.tsx

@@ -21,6 +21,7 @@ export function SourcesCanvas({ welcomeProjectName = null }: SourcesCanvasProps)
     setSelectedDocumentId,
     setCanvasMode,
     handleUploadFile,
+    loadDocuments,
     handleDeleteDocument,
     handleReindexDocument,
     apiStatusLabel,
@@ -39,18 +40,24 @@ export function SourcesCanvas({ welcomeProjectName = null }: SourcesCanvasProps)
       const list = Array.from(files)
       setUploadError(null)
       setUploadingCount(list.length)
+      const isBatch = list.length > 1
       let lastError: string | null = null
       for (const file of list) {
         try {
-          await handleUploadFile(file)
+          await handleUploadFile(file, isBatch ? { skipListRefresh: true } : undefined)
         } catch (e) {
           lastError = e instanceof Error ? e.message : 'Upload failed'
         }
       }
+      if (isBatch) {
+        try {
+          await loadDocuments()
+        } catch {}
+      }
       setUploadingCount(0)
       if (lastError) setUploadError(lastError)
     },
-    [handleUploadFile]
+    [handleUploadFile, loadDocuments]
   )
 
   const handleUploadChange = (e: React.ChangeEvent<HTMLInputElement>) => {

demo/web/src/context/WorkspaceContext.tsx

@@ -22,7 +22,7 @@ interface WorkspaceContextValue {
   setSelectedDocumentId: (id: string | null) => void
   loadProjects: () => Promise<void>
   loadDocuments: () => Promise<void>
-  handleUploadFile: (file: File) => Promise<void>
+  handleUploadFile: (file: File, options?: { skipListRefresh?: boolean }) => Promise<void>
   handleDeleteDocument: (documentId: string) => Promise<void>
   handleReindexDocument: (documentId: string) => Promise<void>
   status: ApiStatus | null
@@ -240,9 +240,10 @@ export function WorkspaceProvider({ children }: WorkspaceProviderProps) {
   )
 
   const handleUploadFile = useCallback(
-    async (file: File) => {
+    async (file: File, options?: { skipListRefresh?: boolean }) => {
       try {
         await uploadDocument(file, currentProjectId)
+        if (options?.skipListRefresh) return
         const docs = await listDocuments(currentProjectId)
         setDocuments(docs)
         if (!selectedDocumentId && docs.length > 0) {