finanzen/convex/imports.ts

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 };
}