feat: added periodic tasks in backend db and frontend hooks

- TODO: Add celery redbeat and create tasks dinamically in our redis
This commit is contained in:
DESKTOP-RTLN3BA\$punk 2025-10-22 16:14:25 -07:00
parent 70808eb08b
commit 182f815bb7
8 changed files with 484 additions and 31 deletions

View file

@ -136,14 +136,6 @@ Check out our public roadmap and contribute your ideas or feedback:
**View the Roadmap:** [SurfSense Roadmap on GitHub Projects](https://github.com/users/MODSetter/projects/2) **View the Roadmap:** [SurfSense Roadmap on GitHub Projects](https://github.com/users/MODSetter/projects/2)
## ⚠️ Important Announcement
**AWS and Vercel are currently experiencing outages.** We deployed a major update to SurfSense last night and have updated our documentation accordingly with important setup and configuration changes. Unfortunately, these documentation updates cannot be deployed to our main site (surfsense.com) due to the ongoing outages.
**Please view our documentation directly on GitHub:**
📚 [SurfSense Documentation](https://github.com/MODSetter/SurfSense/tree/main/surfsense_web/content/docs)
We apologize for any inconvenience and appreciate your patience!
## How to get started? ## How to get started?

View file

@ -55,6 +55,8 @@ def upgrade() -> None:
# ===== STEP 2: Populate search_space_id with user's first search space ===== # ===== STEP 2: Populate search_space_id with user's first search space =====
# This ensures existing LLM configs are assigned to a valid search space # This ensures existing LLM configs are assigned to a valid search space
# Only run this if user_id column exists on llm_configs
if "user_id" in llm_config_columns:
op.execute( op.execute(
""" """
UPDATE llm_configs lc UPDATE llm_configs lc
@ -70,15 +72,28 @@ def upgrade() -> None:
) )
# ===== STEP 3: Make search_space_id NOT NULL and add FK constraint ===== # ===== STEP 3: Make search_space_id NOT NULL and add FK constraint =====
# Check if there are any rows with NULL search_space_id
# If llm_configs table is empty or all rows have search_space_id, we can proceed
result = conn.execute(
sa.text("SELECT COUNT(*) FROM llm_configs WHERE search_space_id IS NULL")
)
null_count = result.scalar()
if null_count == 0 or "user_id" in llm_config_columns:
# Safe to make NOT NULL
op.alter_column( op.alter_column(
"llm_configs", "llm_configs",
"search_space_id", "search_space_id",
nullable=False, nullable=False,
) )
else:
# If there are NULL values and no user_id to migrate from, skip making it NOT NULL
# This would happen if llm_configs already exists without user_id
pass
# Add foreign key constraint # Add foreign key constraint only if search_space_id is NOT NULL
foreign_keys = [fk["name"] for fk in inspector.get_foreign_keys("llm_configs")] foreign_keys = [fk["name"] for fk in inspector.get_foreign_keys("llm_configs")]
if "fk_llm_configs_search_space_id" not in foreign_keys: if "fk_llm_configs_search_space_id" not in foreign_keys and null_count == 0:
op.create_foreign_key( op.create_foreign_key(
"fk_llm_configs_search_space_id", "fk_llm_configs_search_space_id",
"llm_configs", "llm_configs",

View file

@ -0,0 +1,94 @@
"""Add periodic indexing fields to search_source_connectors
Revision ID: 32
Revises: 31
Changes:
1. Add periodic_indexing_enabled column (Boolean, default False)
2. Add indexing_frequency_minutes column (Integer, nullable)
3. Add next_scheduled_at column (TIMESTAMP with timezone, nullable)
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "32"
down_revision: str | None = "31"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Add periodic indexing fields to search_source_connectors table."""
from sqlalchemy import inspect
conn = op.get_bind()
inspector = inspect(conn)
# Get existing columns
connector_columns = [
col["name"] for col in inspector.get_columns("search_source_connectors")
]
# Add periodic_indexing_enabled column if it doesn't exist
if "periodic_indexing_enabled" not in connector_columns:
op.add_column(
"search_source_connectors",
sa.Column(
"periodic_indexing_enabled",
sa.Boolean(),
nullable=False,
server_default="false",
),
)
# Add indexing_frequency_minutes column if it doesn't exist
if "indexing_frequency_minutes" not in connector_columns:
op.add_column(
"search_source_connectors",
sa.Column(
"indexing_frequency_minutes",
sa.Integer(),
nullable=True,
),
)
# Add next_scheduled_at column if it doesn't exist
if "next_scheduled_at" not in connector_columns:
op.add_column(
"search_source_connectors",
sa.Column(
"next_scheduled_at",
sa.TIMESTAMP(timezone=True),
nullable=True,
),
)
def downgrade() -> None:
"""Remove periodic indexing fields from search_source_connectors table."""
from sqlalchemy import inspect
conn = op.get_bind()
inspector = inspect(conn)
# Get existing columns
connector_columns = [
col["name"] for col in inspector.get_columns("search_source_connectors")
]
# Drop columns if they exist
if "next_scheduled_at" in connector_columns:
op.drop_column("search_source_connectors", "next_scheduled_at")
if "indexing_frequency_minutes" in connector_columns:
op.drop_column("search_source_connectors", "indexing_frequency_minutes")
if "periodic_indexing_enabled" in connector_columns:
op.drop_column("search_source_connectors", "periodic_indexing_enabled")

View file

@ -285,6 +285,11 @@ class SearchSourceConnector(BaseModel, TimestampMixin):
last_indexed_at = Column(TIMESTAMP(timezone=True), nullable=True) last_indexed_at = Column(TIMESTAMP(timezone=True), nullable=True)
config = Column(JSON, nullable=False) config = Column(JSON, nullable=False)
# Periodic indexing fields
periodic_indexing_enabled = Column(Boolean, nullable=False, default=False)
indexing_frequency_minutes = Column(Integer, nullable=True)
next_scheduled_at = Column(TIMESTAMP(timezone=True), nullable=True)
search_space_id = Column( search_space_id = Column(
Integer, ForeignKey("searchspaces.id", ondelete="CASCADE"), nullable=False Integer, ForeignKey("searchspaces.id", ondelete="CASCADE"), nullable=False
) )

View file

@ -11,7 +11,7 @@ Note: Each search space can have only one connector of each type per user (based
""" """
import logging import logging
from datetime import datetime, timedelta from datetime import UTC, datetime, timedelta
from typing import Any from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query from fastapi import APIRouter, Depends, HTTPException, Query
@ -124,8 +124,22 @@ async def create_search_source_connector(
status_code=409, status_code=409,
detail=f"A connector with type {connector.connector_type} already exists in this search space. Each search space can have only one connector of each type per user.", detail=f"A connector with type {connector.connector_type} already exists in this search space. Each search space can have only one connector of each type per user.",
) )
# Prepare connector data
connector_data = connector.model_dump()
# Automatically set next_scheduled_at if periodic indexing is enabled
if (
connector.periodic_indexing_enabled
and connector.indexing_frequency_minutes
and connector.next_scheduled_at is None
):
connector_data["next_scheduled_at"] = datetime.now(UTC) + timedelta(
minutes=connector.indexing_frequency_minutes
)
db_connector = SearchSourceConnector( db_connector = SearchSourceConnector(
**connector.model_dump(), search_space_id=search_space_id, user_id=user.id **connector_data, search_space_id=search_space_id, user_id=user.id
) )
session.add(db_connector) session.add(db_connector)
await session.commit() await session.commit()
@ -224,6 +238,50 @@ async def update_search_source_connector(
# Convert the sparse update data (only fields present in request) to a dict # Convert the sparse update data (only fields present in request) to a dict
update_data = connector_update.model_dump(exclude_unset=True) update_data = connector_update.model_dump(exclude_unset=True)
# Validate periodic indexing fields
# Get the effective values after update
effective_is_indexable = update_data.get("is_indexable", db_connector.is_indexable)
effective_periodic_enabled = update_data.get(
"periodic_indexing_enabled", db_connector.periodic_indexing_enabled
)
effective_frequency = update_data.get(
"indexing_frequency_minutes", db_connector.indexing_frequency_minutes
)
# Validate periodic indexing configuration
if effective_periodic_enabled:
if not effective_is_indexable:
raise HTTPException(
status_code=422,
detail="periodic_indexing_enabled can only be True for indexable connectors",
)
if effective_frequency is None:
raise HTTPException(
status_code=422,
detail="indexing_frequency_minutes is required when periodic_indexing_enabled is True",
)
if effective_frequency <= 0:
raise HTTPException(
status_code=422,
detail="indexing_frequency_minutes must be greater than 0",
)
# Automatically set next_scheduled_at if not provided and periodic indexing is being enabled
if (
"periodic_indexing_enabled" in update_data
or "indexing_frequency_minutes" in update_data
) and "next_scheduled_at" not in update_data:
# Schedule the next indexing based on the frequency
update_data["next_scheduled_at"] = datetime.now(UTC) + timedelta(
minutes=effective_frequency
)
elif (
effective_periodic_enabled is False
and "periodic_indexing_enabled" in update_data
):
# If disabling periodic indexing, clear the next_scheduled_at
update_data["next_scheduled_at"] = None
# Special handling for 'config' field # Special handling for 'config' field
if "config" in update_data: if "config" in update_data:
incoming_config = update_data["config"] # Config data from the request incoming_config = update_data["config"] # Config data from the request

View file

@ -2,7 +2,7 @@ import uuid
from datetime import datetime from datetime import datetime
from typing import Any from typing import Any
from pydantic import BaseModel, ConfigDict, field_validator from pydantic import BaseModel, ConfigDict, field_validator, model_validator
from app.db import SearchSourceConnectorType from app.db import SearchSourceConnectorType
from app.utils.validators import validate_connector_config from app.utils.validators import validate_connector_config
@ -16,6 +16,9 @@ class SearchSourceConnectorBase(BaseModel):
is_indexable: bool is_indexable: bool
last_indexed_at: datetime | None = None last_indexed_at: datetime | None = None
config: dict[str, Any] config: dict[str, Any]
periodic_indexing_enabled: bool = False
indexing_frequency_minutes: int | None = None
next_scheduled_at: datetime | None = None
@field_validator("config") @field_validator("config")
@classmethod @classmethod
@ -25,6 +28,22 @@ class SearchSourceConnectorBase(BaseModel):
connector_type = values.data.get("connector_type") connector_type = values.data.get("connector_type")
return validate_connector_config(connector_type, config) return validate_connector_config(connector_type, config)
@model_validator(mode="after")
def validate_periodic_indexing(self):
"""Validate that periodic indexing configuration is consistent."""
if self.periodic_indexing_enabled:
if not self.is_indexable:
raise ValueError(
"periodic_indexing_enabled can only be True for indexable connectors"
)
if self.indexing_frequency_minutes is None:
raise ValueError(
"indexing_frequency_minutes is required when periodic_indexing_enabled is True"
)
if self.indexing_frequency_minutes <= 0:
raise ValueError("indexing_frequency_minutes must be greater than 0")
return self
class SearchSourceConnectorCreate(SearchSourceConnectorBase): class SearchSourceConnectorCreate(SearchSourceConnectorBase):
pass pass
@ -36,6 +55,9 @@ class SearchSourceConnectorUpdate(BaseModel):
is_indexable: bool | None = None is_indexable: bool | None = None
last_indexed_at: datetime | None = None last_indexed_at: datetime | None = None
config: dict[str, Any] | None = None config: dict[str, Any] | None = None
periodic_indexing_enabled: bool | None = None
indexing_frequency_minutes: int | None = None
next_scheduled_at: datetime | None = None
class SearchSourceConnectorRead(SearchSourceConnectorBase, IDModel, TimestampModel): class SearchSourceConnectorRead(SearchSourceConnectorBase, IDModel, TimestampModel):

View file

@ -1,7 +1,15 @@
"use client"; "use client";
import { format } from "date-fns"; import { format } from "date-fns";
import { Calendar as CalendarIcon, Edit, Plus, RefreshCw, Trash2 } from "lucide-react"; import {
Calendar as CalendarIcon,
Clock,
Edit,
Loader2,
Plus,
RefreshCw,
Trash2,
} from "lucide-react";
import { motion } from "motion/react"; import { motion } from "motion/react";
import { useParams, useRouter } from "next/navigation"; import { useParams, useRouter } from "next/navigation";
import { useEffect, useState } from "react"; import { useEffect, useState } from "react";
@ -28,8 +36,17 @@ import {
DialogHeader, DialogHeader,
DialogTitle, DialogTitle,
} from "@/components/ui/dialog"; } from "@/components/ui/dialog";
import { Input } from "@/components/ui/input";
import { Label } from "@/components/ui/label"; import { Label } from "@/components/ui/label";
import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover"; import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover";
import {
Select,
SelectContent,
SelectItem,
SelectTrigger,
SelectValue,
} from "@/components/ui/select";
import { Switch } from "@/components/ui/switch";
import { import {
Table, Table,
TableBody, TableBody,
@ -64,7 +81,7 @@ export default function ConnectorsPage() {
const searchSpaceId = params.search_space_id as string; const searchSpaceId = params.search_space_id as string;
const today = new Date(); const today = new Date();
const { connectors, isLoading, error, deleteConnector, indexConnector } = const { connectors, isLoading, error, deleteConnector, indexConnector, updateConnector } =
useSearchSourceConnectors(false, parseInt(searchSpaceId)); useSearchSourceConnectors(false, parseInt(searchSpaceId));
const [connectorToDelete, setConnectorToDelete] = useState<number | null>(null); const [connectorToDelete, setConnectorToDelete] = useState<number | null>(null);
const [indexingConnectorId, setIndexingConnectorId] = useState<number | null>(null); const [indexingConnectorId, setIndexingConnectorId] = useState<number | null>(null);
@ -75,6 +92,16 @@ export default function ConnectorsPage() {
const [startDate, setStartDate] = useState<Date | undefined>(undefined); const [startDate, setStartDate] = useState<Date | undefined>(undefined);
const [endDate, setEndDate] = useState<Date | undefined>(undefined); const [endDate, setEndDate] = useState<Date | undefined>(undefined);
// Periodic indexing state
const [periodicDialogOpen, setPeriodicDialogOpen] = useState(false);
const [selectedConnectorForPeriodic, setSelectedConnectorForPeriodic] = useState<number | null>(
null
);
const [periodicEnabled, setPeriodicEnabled] = useState(false);
const [frequencyMinutes, setFrequencyMinutes] = useState<string>("1440");
const [customFrequency, setCustomFrequency] = useState<string>("");
const [isSavingPeriodic, setIsSavingPeriodic] = useState(false);
useEffect(() => { useEffect(() => {
if (error) { if (error) {
toast.error("Failed to load connectors"); toast.error("Failed to load connectors");
@ -141,6 +168,84 @@ export default function ConnectorsPage() {
} }
}; };
// Handle opening periodic indexing dialog
const handleOpenPeriodicDialog = (connectorId: number) => {
const connector = connectors.find((c) => c.id === connectorId);
if (!connector) return;
setSelectedConnectorForPeriodic(connectorId);
setPeriodicEnabled(connector.periodic_indexing_enabled);
if (connector.indexing_frequency_minutes) {
// Check if it's a preset value
const presetValues = ["15", "60", "360", "720", "1440", "10080"];
if (presetValues.includes(connector.indexing_frequency_minutes.toString())) {
setFrequencyMinutes(connector.indexing_frequency_minutes.toString());
setCustomFrequency("");
} else {
setFrequencyMinutes("custom");
setCustomFrequency(connector.indexing_frequency_minutes.toString());
}
} else {
setFrequencyMinutes("1440");
setCustomFrequency("");
}
setPeriodicDialogOpen(true);
};
// Handle saving periodic indexing configuration
const handleSavePeriodicIndexing = async () => {
if (selectedConnectorForPeriodic === null) return;
const connector = connectors.find((c) => c.id === selectedConnectorForPeriodic);
if (!connector) return;
setIsSavingPeriodic(true);
try {
// Determine the frequency value
let frequency: number | null = null;
if (periodicEnabled) {
if (frequencyMinutes === "custom") {
frequency = parseInt(customFrequency, 10);
if (isNaN(frequency) || frequency <= 0) {
toast.error("Please enter a valid frequency in minutes");
setIsSavingPeriodic(false);
return;
}
} else {
frequency = parseInt(frequencyMinutes, 10);
}
}
await updateConnector(selectedConnectorForPeriodic, {
periodic_indexing_enabled: periodicEnabled,
indexing_frequency_minutes: frequency,
});
toast.success(
periodicEnabled
? "Periodic indexing enabled successfully"
: "Periodic indexing disabled successfully"
);
setPeriodicDialogOpen(false);
} catch (error) {
console.error("Error updating periodic indexing:", error);
toast.error(error instanceof Error ? error.message : "Failed to update periodic indexing");
} finally {
setIsSavingPeriodic(false);
setSelectedConnectorForPeriodic(null);
}
};
// Format frequency for display
const formatFrequency = (minutes: number): string => {
if (minutes < 60) return `${minutes}m`;
if (minutes < 1440) return `${Math.floor(minutes / 60)}h`;
if (minutes < 10080) return `${Math.floor(minutes / 1440)}d`;
return `${Math.floor(minutes / 10080)}w`;
};
return ( return (
<div className="container mx-auto py-8 max-w-6xl"> <div className="container mx-auto py-8 max-w-6xl">
<motion.div <motion.div
@ -193,6 +298,7 @@ export default function ConnectorsPage() {
<TableHead>Name</TableHead> <TableHead>Name</TableHead>
<TableHead>Type</TableHead> <TableHead>Type</TableHead>
<TableHead>Last Indexed</TableHead> <TableHead>Last Indexed</TableHead>
<TableHead>Periodic</TableHead>
<TableHead className="text-right">Actions</TableHead> <TableHead className="text-right">Actions</TableHead>
</TableRow> </TableRow>
</TableHeader> </TableHeader>
@ -206,6 +312,41 @@ export default function ConnectorsPage() {
? formatDateTime(connector.last_indexed_at) ? formatDateTime(connector.last_indexed_at)
: "Not indexable"} : "Not indexable"}
</TableCell> </TableCell>
<TableCell>
{connector.is_indexable ? (
connector.periodic_indexing_enabled ? (
<TooltipProvider>
<Tooltip>
<TooltipTrigger asChild>
<div className="flex items-center gap-1 text-green-600 dark:text-green-400">
<Clock className="h-4 w-4" />
<span className="text-sm font-medium">
{connector.indexing_frequency_minutes
? formatFrequency(connector.indexing_frequency_minutes)
: "Enabled"}
</span>
</div>
</TooltipTrigger>
<TooltipContent>
<p>
Runs every {connector.indexing_frequency_minutes} minutes
{connector.next_scheduled_at && (
<>
<br />
Next: {formatDateTime(connector.next_scheduled_at)}
</>
)}
</p>
</TooltipContent>
</Tooltip>
</TooltipProvider>
) : (
<span className="text-sm text-muted-foreground">Disabled</span>
)
) : (
<span className="text-sm text-muted-foreground">-</span>
)}
</TableCell>
<TableCell className="text-right"> <TableCell className="text-right">
<div className="flex justify-end gap-2"> <div className="flex justify-end gap-2">
{connector.is_indexable && ( {connector.is_indexable && (
@ -256,6 +397,25 @@ export default function ConnectorsPage() {
</TooltipProvider> </TooltipProvider>
</div> </div>
)} )}
{connector.is_indexable && (
<TooltipProvider>
<Tooltip>
<TooltipTrigger asChild>
<Button
variant="outline"
size="sm"
onClick={() => handleOpenPeriodicDialog(connector.id)}
>
<Clock className="h-4 w-4" />
<span className="sr-only">Configure Periodic Indexing</span>
</Button>
</TooltipTrigger>
<TooltipContent>
<p>Configure Periodic Indexing</p>
</TooltipContent>
</Tooltip>
</TooltipProvider>
)}
<Button <Button
variant="outline" variant="outline"
size="sm" size="sm"
@ -424,6 +584,110 @@ export default function ConnectorsPage() {
</DialogFooter> </DialogFooter>
</DialogContent> </DialogContent>
</Dialog> </Dialog>
{/* Periodic Indexing Configuration Dialog */}
<Dialog open={periodicDialogOpen} onOpenChange={setPeriodicDialogOpen}>
<DialogContent className="sm:max-w-[500px]">
<DialogHeader>
<DialogTitle>Configure Periodic Indexing</DialogTitle>
<DialogDescription>
Set up automatic indexing at regular intervals for this connector.
</DialogDescription>
</DialogHeader>
<div className="grid gap-6 py-4">
<div className="flex items-center justify-between space-x-2">
<div className="space-y-0.5">
<Label htmlFor="periodic-enabled" className="text-base">
Enable Periodic Indexing
</Label>
<p className="text-sm text-muted-foreground">
Automatically index this connector at regular intervals
</p>
</div>
<Switch
id="periodic-enabled"
checked={periodicEnabled}
onCheckedChange={setPeriodicEnabled}
/>
</div>
{periodicEnabled && (
<div className="space-y-4">
<div className="space-y-2">
<Label htmlFor="frequency">Indexing Frequency</Label>
<Select value={frequencyMinutes} onValueChange={setFrequencyMinutes}>
<SelectTrigger id="frequency">
<SelectValue placeholder="Select frequency" />
</SelectTrigger>
<SelectContent>
<SelectItem value="15">Every 15 minutes</SelectItem>
<SelectItem value="60">Every hour</SelectItem>
<SelectItem value="360">Every 6 hours</SelectItem>
<SelectItem value="720">Every 12 hours</SelectItem>
<SelectItem value="1440">Daily (24 hours)</SelectItem>
<SelectItem value="10080">Weekly (7 days)</SelectItem>
<SelectItem value="custom">Custom</SelectItem>
</SelectContent>
</Select>
</div>
{frequencyMinutes === "custom" && (
<div className="space-y-2">
<Label htmlFor="custom-frequency">Custom Frequency (minutes)</Label>
<Input
id="custom-frequency"
type="number"
min="1"
placeholder="Enter minutes"
value={customFrequency}
onChange={(e) => setCustomFrequency(e.target.value)}
/>
<p className="text-xs text-muted-foreground">
Enter the number of minutes between each indexing run
</p>
</div>
)}
<div className="rounded-lg bg-muted p-3 text-sm">
<p className="font-medium mb-1">Preview:</p>
<p className="text-muted-foreground">
{frequencyMinutes === "custom" && customFrequency
? `Will run every ${customFrequency} minutes`
: frequencyMinutes === "15"
? "Will run every 15 minutes"
: frequencyMinutes === "60"
? "Will run every hour"
: frequencyMinutes === "360"
? "Will run every 6 hours"
: frequencyMinutes === "720"
? "Will run every 12 hours"
: frequencyMinutes === "1440"
? "Will run daily (every 24 hours)"
: frequencyMinutes === "10080"
? "Will run weekly (every 7 days)"
: "Select a frequency above"}
</p>
</div>
</div>
)}
</div>
<DialogFooter>
<Button
variant="outline"
onClick={() => {
setPeriodicDialogOpen(false);
setSelectedConnectorForPeriodic(null);
}}
>
Cancel
</Button>
<Button onClick={handleSavePeriodicIndexing} disabled={isSavingPeriodic}>
{isSavingPeriodic && <Loader2 className="mr-2 h-4 w-4 animate-spin" />}
Save Configuration
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
</div> </div>
); );
} }

View file

@ -10,6 +10,9 @@ export interface SearchSourceConnector {
search_space_id: number; search_space_id: number;
user_id?: string; user_id?: string;
created_at?: string; created_at?: string;
periodic_indexing_enabled: boolean;
indexing_frequency_minutes: number | null;
next_scheduled_at: string | null;
} }
export interface ConnectorSourceItem { export interface ConnectorSourceItem {