feat: Implement analysis job tracking with progress timeline and enhanced data source status management.

convex/dataSources.ts

@@ -19,6 +19,22 @@ export const getProjectDataSources = query({
     },
 });
 
+export const getById = query({
+    args: { dataSourceId: v.id("dataSources") },
+    handler: async (ctx, args) => {
+        const userId = await getAuthUserId(ctx);
+        if (!userId) return null;
+
+        const dataSource = await ctx.db.get(args.dataSourceId);
+        if (!dataSource) return null;
+
+        const project = await ctx.db.get(dataSource.projectId);
+        if (!project || project.userId !== userId) return null;
+
+        return dataSource;
+    },
+});
+
 export const addDataSource = mutation({
     args: {
         projectId: v.optional(v.id("projects")), // Optional, if not provided, use default
@@ -30,6 +46,14 @@ export const addDataSource = mutation({
         const userId = await getAuthUserId(ctx);
         if (!userId) throw new Error("Unauthorized");
 
+        let normalizedUrl = args.url.trim();
+        if (normalizedUrl.startsWith("manual:")) {
+            // Keep manual sources as-is.
+        } else if (!normalizedUrl.startsWith("http")) {
+            normalizedUrl = `https://${normalizedUrl}`;
+        }
+        normalizedUrl = normalizedUrl.replace(/\/+$/, "");
+
         let projectId = args.projectId;
 
         // Use default project if not provided
@@ -55,7 +79,7 @@ export const addDataSource = mutation({
         const existing = await ctx.db
             .query("dataSources")
             .withIndex("by_project_url", (q) =>
-                q.eq("projectId", projectId!).eq("url", args.url)
+                q.eq("projectId", projectId!).eq("url", normalizedUrl)
             )
             .first();
 
@@ -64,7 +88,7 @@ export const addDataSource = mutation({
             : await ctx.db.insert("dataSources", {
                   projectId: projectId!, // Assert exists
                   type: args.type,
-                  url: args.url,
+                  url: normalizedUrl,
                   name: args.name,
                   analysisStatus: "pending",
                   lastAnalyzedAt: undefined,
@@ -83,7 +107,7 @@ export const addDataSource = mutation({
             }
         }
 
-        return { sourceId, projectId: projectId! };
+        return { sourceId, projectId: projectId!, isExisting: Boolean(existing) };
     },
 });
 
@@ -117,3 +141,46 @@ export const updateDataSourceStatus = mutation({
         });
     },
 });
+
+export const remove = mutation({
+    args: { dataSourceId: v.id("dataSources") },
+    handler: async (ctx, args) => {
+        const userId = await getAuthUserId(ctx);
+        if (!userId) throw new Error("Unauthorized");
+
+        const dataSource = await ctx.db.get(args.dataSourceId);
+        if (!dataSource) throw new Error("Data source not found");
+
+        const project = await ctx.db.get(dataSource.projectId);
+        if (!project || project.userId !== userId) {
+            throw new Error("Project not found or unauthorized");
+        }
+
+        const updatedSelected = project.dorkingConfig.selectedSourceIds.filter(
+            (id) => id !== args.dataSourceId
+        );
+        await ctx.db.patch(project._id, {
+            dorkingConfig: { selectedSourceIds: updatedSelected },
+        });
+
+        const analyses = await ctx.db
+            .query("analyses")
+            .withIndex("by_dataSource_createdAt", (q) =>
+                q.eq("dataSourceId", args.dataSourceId)
+            )
+            .collect();
+        for (const analysis of analyses) {
+            await ctx.db.delete(analysis._id);
+        }
+
+        const analysisJobs = await ctx.db
+            .query("analysisJobs")
+            .filter((q) => q.eq(q.field("dataSourceId"), args.dataSourceId))
+            .collect();
+        for (const job of analysisJobs) {
+            await ctx.db.delete(job._id);
+        }
+
+        await ctx.db.delete(args.dataSourceId);
+    },
+});