initial commit

convex/imports.ts

@@ -0,0 +1,208 @@
+import { query, mutation, internalMutation } from "./_generated/server";
+import { v } from "convex/values";
+import type { MutationCtx } from "./_generated/server";
+import type { Id } from "./_generated/dataModel";
+import {
+  assertOwned,
+  enrichTransactionFields,
+  requireUserId,
+} from "./lib/helpers";
+
+const importRowValidator = v.object({
+  accountId: v.optional(v.id("accounts")),
+  categoryId: v.optional(v.id("categories")),
+  categoryName: v.optional(v.string()),
+  bookingDate: v.optional(v.string()),
+  valueDate: v.optional(v.string()),
+  description: v.string(),
+  counterparty: v.optional(v.string()),
+  amount: v.number(),
+  vorgang: v.optional(v.string()),
+  isPending: v.boolean(),
+  notes: v.optional(v.string()),
+  rawText: v.optional(v.string()),
+  assignedMonth: v.optional(v.string()),
+  effectiveMonth: v.optional(v.string()),
+  dedupHash: v.optional(v.string()),
+  externalRef: v.optional(v.string()),
+});
+
+export const list = query({
+  args: {},
+  returns: v.array(
+    v.object({
+      _id: v.id("imports"),
+      _creationTime: v.number(),
+      userId: v.id("users"),
+      filename: v.string(),
+      source: v.string(),
+      accountId: v.optional(v.id("accounts")),
+      rowCount: v.number(),
+      importedCount: v.number(),
+      status: v.string(),
+    }),
+  ),
+  handler: async (ctx) => {
+    const userId = await requireUserId(ctx);
+    return await ctx.db
+      .query("imports")
+      .withIndex("by_user", (q) => q.eq("userId", userId))
+      .order("desc")
+      .collect();
+  },
+});
+
+export const commitRows = mutation({
+  args: {
+    filename: v.string(),
+    source: v.string(),
+    accountId: v.optional(v.id("accounts")),
+    rows: v.array(importRowValidator),
+  },
+  returns: v.object({
+    importId: v.id("imports"),
+    importedCount: v.number(),
+    skippedCount: v.number(),
+  }),
+  handler: async (ctx, args) => {
+    const userId = await requireUserId(ctx);
+    return await commitRowsHandler(ctx, userId, args);
+  },
+});
+
+export const commitRowsInternal = internalMutation({
+  args: {
+    userId: v.id("users"),
+    filename: v.string(),
+    source: v.string(),
+    accountId: v.optional(v.id("accounts")),
+    rows: v.array(importRowValidator),
+  },
+  returns: v.object({
+    importId: v.id("imports"),
+    importedCount: v.number(),
+    skippedCount: v.number(),
+  }),
+  handler: async (ctx, args) => {
+    const { userId, ...rest } = args;
+    return await commitRowsHandler(ctx, userId, rest);
+  },
+});
+
+async function commitRowsHandler(
+  ctx: MutationCtx,
+  userId: Id<"users">,
+  args: {
+    filename: string;
+    source: string;
+    accountId?: Id<"accounts">;
+    rows: Array<{
+      accountId?: Id<"accounts">;
+      categoryId?: Id<"categories">;
+      categoryName?: string;
+      bookingDate?: string;
+      valueDate?: string;
+      description: string;
+      counterparty?: string;
+      amount: number;
+      vorgang?: string;
+      isPending: boolean;
+      notes?: string;
+      rawText?: string;
+      assignedMonth?: string;
+      effectiveMonth?: string;
+      dedupHash?: string;
+      externalRef?: string;
+    }>;
+  },
+) {
+    if (args.accountId) {
+      await assertOwned(await ctx.db.get("accounts", args.accountId), userId, "Konto");
+    }
+
+    const importId = await ctx.db.insert("imports", {
+      userId,
+      filename: args.filename,
+      source: args.source,
+      accountId: args.accountId,
+      rowCount: args.rows.length,
+      importedCount: 0,
+      status: "processing",
+    });
+
+    let importedCount = 0;
+    let skippedCount = 0;
+
+    for (const row of args.rows) {
+      if (row.externalRef) {
+        const existingRef = await ctx.db
+          .query("transactions")
+          .withIndex("by_user_extref", (q) =>
+            q.eq("userId", userId).eq("externalRef", row.externalRef),
+          )
+          .unique();
+        if (existingRef) {
+          skippedCount++;
+          continue;
+        }
+      }
+
+      const enriched = await enrichTransactionFields(ctx, userId, {
+        ...row,
+        accountId: row.accountId ?? args.accountId,
+        importId,
+      });
+
+      const dedupHash = row.dedupHash ?? enriched.dedupHash;
+      const existingDedup = await ctx.db
+        .query("transactions")
+        .withIndex("by_user_dedup", (q) =>
+          q.eq("userId", userId).eq("dedupHash", dedupHash),
+        )
+        .unique();
+      if (existingDedup) {
+        skippedCount++;
+        continue;
+      }
+
+      let categoryId = row.categoryId;
+      if (!categoryId && row.categoryName) {
+        const cat = await ctx.db
+          .query("categories")
+          .withIndex("by_user_name", (q) =>
+            q.eq("userId", userId).eq("name", row.categoryName!),
+          )
+          .unique();
+        categoryId = cat?._id;
+      }
+      if (!categoryId) categoryId = enriched.categoryId;
+
+      await ctx.db.insert("transactions", {
+        userId,
+        accountId: row.accountId ?? args.accountId,
+        categoryId,
+        bookingDate: row.isPending ? undefined : row.bookingDate,
+        valueDate: row.valueDate,
+        description: row.description,
+        counterparty: row.counterparty,
+        amount: enriched.amount,
+        vorgang: row.vorgang,
+        isPending: row.isPending,
+        notes: row.notes,
+        rawText: row.rawText,
+        importId,
+        assignedMonth: row.assignedMonth ?? enriched.assignedMonth,
+        effectiveMonth: row.effectiveMonth ?? enriched.effectiveMonth,
+        dedupHash,
+        externalRef: row.externalRef,
+      });
+      importedCount++;
+    }
+
+    await ctx.db.patch(importId, {
+      importedCount,
+      status: "completed",
+    });
+
+    return { importId, importedCount, skippedCount };
+}