bump

convex/importer.ts

@@ -0,0 +1,677 @@
+import { query, internalMutation } from "./_generated/server";
+import { v } from "convex/values";
+import type { Id } from "./_generated/dataModel";
+import { hasPermission } from "./roles";
+import { AUDIT_ACTIONS, logAudit } from "./audit";
+import { getPublicStorageUrl } from "./storageUrl";
+
+/**
+ * Discord backup importer — server-side non-node half.
+ *
+ * Ghost profiles
+ * --------------
+ * A ghost is a `userProfiles` row with `isGhost: true` and random
+ * junk in every auth-sensitive field. It can't log in (its DAK hash
+ * can't be reproduced), doesn't get channelKeys / roles / presence,
+ * and is filtered out of member / mention / DM-target lookups. The
+ * admin creates ghosts so imported messages have a real `senderId`
+ * to point at; later, "merge" replaces every imported message's
+ * senderId with a real user's ID and deletes the ghost.
+ *
+ * All public entry points live in `importerActions.ts` because they
+ * run node crypto to verify the admin's Ed25519 signature.
+ */
+
+const MAX_CIPHERTEXT_CHARS = 64 * 1024;
+const MAX_IMPORT_BATCH = 100;
+const MERGE_PAGE = 200;
+
+/**
+ * Upsert a batch of ghost profiles keyed by Discord snowflake.
+ *
+ * Idempotent: re-running for the same `discordId` returns the
+ * existing row. If the incoming `displayName` / `avatarUrl` changes,
+ * the existing ghost is patched so the UI reflects the most recent
+ * Discord snapshot — but a real user who already had `discordId`
+ * attached (via a prior merge) is never touched.
+ *
+ * Internal-only. Called from `importerActions.prepareGhostsAction`
+ * which does the signature + admin-permission check.
+ */
+export const ensureGhostsInternal = internalMutation({
+  args: {
+    authors: v.array(
+      v.object({
+        discordId: v.string(),
+        username: v.string(),
+        displayName: v.optional(v.string()),
+        avatarUrl: v.optional(v.string()),
+      }),
+    ),
+  },
+  returns: v.array(
+    v.object({
+      discordId: v.string(),
+      userId: v.id("userProfiles"),
+      created: v.boolean(),
+      isGhost: v.boolean(),
+    }),
+  ),
+  handler: async (ctx, args) => {
+    const out: Array<{
+      discordId: string;
+      userId: Id<"userProfiles">;
+      created: boolean;
+      isGhost: boolean;
+    }> = [];
+    for (const author of args.authors) {
+      const existing = await ctx.db
+        .query("userProfiles")
+        .withIndex("by_discord_id", (q) => q.eq("discordId", author.discordId))
+        .first();
+      if (existing) {
+        if (existing.isGhost) {
+          const patch: Record<string, unknown> = {};
+          if (author.displayName && author.displayName !== existing.displayName) {
+            patch.displayName = author.displayName;
+          }
+          if (author.avatarUrl && author.avatarUrl !== existing.ghostAvatarUrl) {
+            patch.ghostAvatarUrl = author.avatarUrl;
+          }
+          if (Object.keys(patch).length > 0) {
+            await ctx.db.patch(existing._id, patch);
+          }
+        }
+        out.push({
+          discordId: author.discordId,
+          userId: existing._id,
+          created: false,
+          isGhost: !!existing.isGhost,
+        });
+        continue;
+      }
+
+      // Ghost usernames are deterministically scoped by Discord
+      // snowflake so they never collide with real usernames, even
+      // if two Discord users happen to share a handle locally.
+      const ghostUsername = `discord_${author.discordId}`;
+      const junk = `ghost-${author.discordId}-${Date.now()}`;
+      const userId = await ctx.db.insert("userProfiles", {
+        username: ghostUsername,
+        clientSalt: junk,
+        encryptedMasterKey: "",
+        hashedAuthKey: junk,
+        publicIdentityKey: "",
+        publicSigningKey: "",
+        encryptedPrivateKeys: "",
+        isAdmin: false,
+        isGhost: true,
+        discordId: author.discordId,
+        displayName: author.displayName ?? author.username,
+        ghostAvatarUrl: author.avatarUrl,
+      });
+      out.push({
+        discordId: author.discordId,
+        userId,
+        created: true,
+        isGhost: true,
+      });
+    }
+    return out;
+  },
+});
+
+/**
+ * Insert a batch of pre-encrypted imported messages. Skips rows
+ * whose `discordMessageId` already exists in the target channel so
+ * re-runs after a partial failure don't duplicate history.
+ *
+ * `isBanned` is intentionally NOT checked: ghosts can't be banned
+ * (no login path) and historical messages from a later-banned real
+ * user shouldn't be blocked from import. The admin-only entry
+ * point is the permission gate.
+ *
+ * Returns the per-row Convex IDs keyed by the original Discord
+ * message ID so the client can resolve cross-batch replies via
+ * `resolveReplyTargets` without a second query.
+ */
+export const importBatchInternal = internalMutation({
+  args: {
+    actorId: v.id("userProfiles"),
+    channelId: v.id("channels"),
+    messages: v.array(
+      v.object({
+        senderId: v.id("userProfiles"),
+        ciphertext: v.string(),
+        nonce: v.string(),
+        signature: v.string(),
+        keyVersion: v.number(),
+        replyTo: v.optional(v.id("messages")),
+        importedCreatedAt: v.number(),
+        discordMessageId: v.string(),
+      }),
+    ),
+  },
+  returns: v.object({
+    inserted: v.number(),
+    skipped: v.number(),
+    resolved: v.array(
+      v.object({
+        discordMessageId: v.string(),
+        messageId: v.id("messages"),
+      }),
+    ),
+  }),
+  handler: async (ctx, args) => {
+    if (!(await hasPermission(ctx, args.actorId, "manage_channels"))) {
+      throw new Error("You don't have permission to import messages.");
+    }
+    if (args.messages.length === 0) {
+      return { inserted: 0, skipped: 0, resolved: [] };
+    }
+    if (args.messages.length > MAX_IMPORT_BATCH) {
+      throw new Error(`Batch too large (max ${MAX_IMPORT_BATCH}).`);
+    }
+
+    let inserted = 0;
+    let skipped = 0;
+    const resolved: Array<{ discordMessageId: string; messageId: Id<"messages"> }> = [];
+
+    for (const msg of args.messages) {
+      if (msg.ciphertext.length > MAX_CIPHERTEXT_CHARS) {
+        throw new Error("Imported message too large");
+      }
+      const existing = await ctx.db
+        .query("messages")
+        .withIndex("by_channel_discord_message_id", (q) =>
+          q
+            .eq("channelId", args.channelId)
+            .eq("discordMessageId", msg.discordMessageId),
+        )
+        .first();
+      if (existing) {
+        skipped++;
+        resolved.push({
+          discordMessageId: msg.discordMessageId,
+          messageId: existing._id,
+        });
+        continue;
+      }
+      const id = await ctx.db.insert("messages", {
+        channelId: args.channelId,
+        senderId: msg.senderId,
+        ciphertext: msg.ciphertext,
+        nonce: msg.nonce,
+        signature: msg.signature,
+        keyVersion: msg.keyVersion,
+        replyTo: msg.replyTo,
+        importedCreatedAt: msg.importedCreatedAt,
+        discordMessageId: msg.discordMessageId,
+        isImported: true,
+      });
+      inserted++;
+      resolved.push({ discordMessageId: msg.discordMessageId, messageId: id });
+    }
+
+    if (inserted > 0) {
+      await logAudit(ctx, {
+        actorId: args.actorId,
+        action: AUDIT_ACTIONS.MESSAGES_IMPORT_BULK,
+        targetType: "channel",
+        targetId: args.channelId,
+        metadata: { inserted, skipped, total: args.messages.length },
+      });
+    }
+    return { inserted, skipped, resolved };
+  },
+});
+
+/**
+ * Merge step 1: rewrite one page of messages from `ghostUserId` to
+ * `targetUserId`. Called repeatedly from the action until no more
+ * pages. Kept at a bounded page size (`MERGE_PAGE`) so a single
+ * mutation never trips Convex's execution limits on users with
+ * tens of thousands of imported messages.
+ *
+ * The action also tracks the count across pages for the audit log.
+ */
+export const mergeGhostPageInternal = internalMutation({
+  args: {
+    actorId: v.id("userProfiles"),
+    ghostUserId: v.id("userProfiles"),
+    targetUserId: v.id("userProfiles"),
+  },
+  returns: v.object({ rewritten: v.number(), done: v.boolean() }),
+  handler: async (ctx, args) => {
+    if (!(await hasPermission(ctx, args.actorId, "manage_channels"))) {
+      throw new Error("You don't have permission to merge ghosts.");
+    }
+    const ghost = await ctx.db.get(args.ghostUserId);
+    if (!ghost) throw new Error("Ghost not found");
+    if (!ghost.isGhost) throw new Error("Refusing to merge a non-ghost user");
+    const target = await ctx.db.get(args.targetUserId);
+    if (!target) throw new Error("Target user not found");
+    if (target.isGhost) throw new Error("Merge target must be a real user");
+
+    const page = await ctx.db
+      .query("messages")
+      .withIndex("by_sender", (q) => q.eq("senderId", args.ghostUserId))
+      .take(MERGE_PAGE);
+    for (const m of page) {
+      await ctx.db.patch(m._id, { senderId: args.targetUserId });
+    }
+    return { rewritten: page.length, done: page.length < MERGE_PAGE };
+  },
+});
+
+/**
+ * Final step of the merge flow: once all messages have been
+ * rewritten, inherit the ghost's `discordId` onto the target (so
+ * future imports of the same Discord author auto-link), delete the
+ * ghost, and write a single audit entry.
+ */
+export const finalizeMergeInternal = internalMutation({
+  args: {
+    actorId: v.id("userProfiles"),
+    ghostUserId: v.id("userProfiles"),
+    targetUserId: v.id("userProfiles"),
+    totalRewritten: v.number(),
+  },
+  returns: v.null(),
+  handler: async (ctx, args) => {
+    if (!(await hasPermission(ctx, args.actorId, "manage_channels"))) {
+      throw new Error("You don't have permission to merge ghosts.");
+    }
+    const ghost = await ctx.db.get(args.ghostUserId);
+    if (!ghost) return null; // already finalized
+    if (!ghost.isGhost) throw new Error("Refusing to finalize a non-ghost user");
+    const target = await ctx.db.get(args.targetUserId);
+    if (!target) throw new Error("Target user not found");
+
+    // Guard against a stray message slipping in between the last
+    // page and this finalize call. If anything's left, refuse —
+    // the caller will run another page.
+    const stray = await ctx.db
+      .query("messages")
+      .withIndex("by_sender", (q) => q.eq("senderId", args.ghostUserId))
+      .first();
+    if (stray) {
+      throw new Error("Messages still attributed to ghost; run another page.");
+    }
+
+    // Inherit discordId if the target doesn't already have one. If
+    // the target already claimed a different Discord identity we
+    // don't overwrite — that's an admin error, not ours to resolve.
+    if (ghost.discordId && !target.discordId) {
+      await ctx.db.patch(args.targetUserId, { discordId: ghost.discordId });
+    }
+
+    await ctx.db.delete(args.ghostUserId);
+
+    await logAudit(ctx, {
+      actorId: args.actorId,
+      action: AUDIT_ACTIONS.GHOST_MERGE,
+      targetType: "user",
+      targetId: args.targetUserId,
+      targetName: target.displayName ?? target.username,
+      metadata: {
+        ghostDisplayName: ghost.displayName,
+        ghostDiscordId: ghost.discordId,
+        rewritten: args.totalRewritten,
+      },
+    });
+    return null;
+  },
+});
+
+/**
+ * Bulk-delete all imported messages in a channel. Paged so even a
+ * channel with tens of thousands of imports completes without
+ * tripping Convex's per-mutation execution limit. The action layer
+ * loops until `done: true`.
+ *
+ * Used by the Import tab's "Clear imports" button to wipe blank
+ * rows left behind by a failed earlier import (eg. partial
+ * attachment uploads under the old silent-skip behaviour) so the
+ * admin can re-run from scratch.
+ */
+const CLEAR_IMPORTS_PAGE = 200;
+
+export const clearChannelImportsPageInternal = internalMutation({
+  args: {
+    actorId: v.id("userProfiles"),
+    channelId: v.id("channels"),
+  },
+  returns: v.object({ deleted: v.number(), done: v.boolean() }),
+  handler: async (ctx, args) => {
+    if (!(await hasPermission(ctx, args.actorId, "manage_channels"))) {
+      throw new Error("You don't have permission to clear imports.");
+    }
+    // `by_channel_imported_at` is keyed [channelId, importedCreatedAt].
+    // Live messages don't set `importedCreatedAt`, so this index is
+    // effectively scoped to imported rows for this channel — no
+    // accidental deletion of live content.
+    const page = await ctx.db
+      .query("messages")
+      .withIndex("by_channel_imported_at", (q) =>
+        q.eq("channelId", args.channelId),
+      )
+      .take(CLEAR_IMPORTS_PAGE);
+    let deleted = 0;
+    for (const m of page) {
+      if (!m.isImported) continue; // defence-in-depth
+      // Cascade: drop any reactions tied to this message so we
+      // don't orphan rows in `messageReactions`.
+      const reactions = await ctx.db
+        .query("messageReactions")
+        .withIndex("by_message", (q) => q.eq("messageId", m._id))
+        .collect();
+      for (const r of reactions) await ctx.db.delete(r._id);
+      await ctx.db.delete(m._id);
+      deleted++;
+    }
+    return { deleted, done: page.length < CLEAR_IMPORTS_PAGE };
+  },
+});
+
+export const finalizeClearImportsInternal = internalMutation({
+  args: {
+    actorId: v.id("userProfiles"),
+    channelId: v.id("channels"),
+    totalDeleted: v.number(),
+  },
+  returns: v.null(),
+  handler: async (ctx, args) => {
+    if (!(await hasPermission(ctx, args.actorId, "manage_channels"))) {
+      throw new Error("You don't have permission to clear imports.");
+    }
+    const channel = await ctx.db.get(args.channelId);
+    await logAudit(ctx, {
+      actorId: args.actorId,
+      action: AUDIT_ACTIONS.MESSAGES_IMPORT_BULK,
+      targetType: "channel",
+      targetId: args.channelId,
+      targetName: channel?.name,
+      metadata: { cleared: args.totalDeleted },
+    });
+    return null;
+  },
+});
+
+/**
+ * Surgical-repair companion to `resolveReplyTargets`. Returns the
+ * full decryptable body (ciphertext + nonce + keyVersion) for each
+ * existing imported message in `discordMessageIds`. The runner
+ * uses this in repair mode to decrypt each row locally and decide
+ * whether its attachment list is missing entries — cheaper than
+ * dropping and re-inserting every row.
+ */
+export const getImportedState = query({
+  args: {
+    channelId: v.id("channels"),
+    discordMessageIds: v.array(v.string()),
+  },
+  returns: v.array(
+    v.object({
+      discordMessageId: v.string(),
+      messageId: v.id("messages"),
+      ciphertext: v.string(),
+      nonce: v.string(),
+      keyVersion: v.number(),
+    }),
+  ),
+  handler: async (ctx, args) => {
+    const out: Array<{
+      discordMessageId: string;
+      messageId: Id<"messages">;
+      ciphertext: string;
+      nonce: string;
+      keyVersion: number;
+    }> = [];
+    for (const dId of args.discordMessageIds) {
+      const row = await ctx.db
+        .query("messages")
+        .withIndex("by_channel_discord_message_id", (q) =>
+          q.eq("channelId", args.channelId).eq("discordMessageId", dId),
+        )
+        .first();
+      if (!row) continue;
+      out.push({
+        discordMessageId: dId,
+        messageId: row._id,
+        ciphertext: row.ciphertext,
+        nonce: row.nonce,
+        keyVersion: row.keyVersion,
+      });
+    }
+    return out;
+  },
+});
+
+/**
+ * Delete a specific list of imported messages by their Discord
+ * snowflakes. Paged so any-size list is safe. Used by the runner's
+ * repair mode to surgically drop rows that decrypted to an empty
+ * or under-attached plaintext, before re-inserting them fresh.
+ */
+const DELETE_BY_DISCORD_PAGE = 100;
+
+export const deleteImportedByDiscordIdsInternal = internalMutation({
+  args: {
+    actorId: v.id("userProfiles"),
+    channelId: v.id("channels"),
+    discordMessageIds: v.array(v.string()),
+  },
+  returns: v.object({ deleted: v.number() }),
+  handler: async (ctx, args) => {
+    if (!(await hasPermission(ctx, args.actorId, "manage_channels"))) {
+      throw new Error("You don't have permission to repair imports.");
+    }
+    if (args.discordMessageIds.length > DELETE_BY_DISCORD_PAGE) {
+      throw new Error(`Delete batch too large (max ${DELETE_BY_DISCORD_PAGE}).`);
+    }
+    let deleted = 0;
+    for (const dId of args.discordMessageIds) {
+      const row = await ctx.db
+        .query("messages")
+        .withIndex("by_channel_discord_message_id", (q) =>
+          q.eq("channelId", args.channelId).eq("discordMessageId", dId),
+        )
+        .first();
+      if (!row) continue;
+      if (!row.isImported) continue; // defence-in-depth
+      const reactions = await ctx.db
+        .query("messageReactions")
+        .withIndex("by_message", (q) => q.eq("messageId", row._id))
+        .collect();
+      for (const r of reactions) await ctx.db.delete(r._id);
+      await ctx.db.delete(row._id);
+      deleted++;
+    }
+    return { deleted };
+  },
+});
+
+/**
+ * Reply-remapping helper. Given a list of Discord message IDs the
+ * import runner has yet to place, return the Convex IDs for the
+ * ones that ARE already imported in this channel. Unimported
+ * entries are omitted (client treats missing keys as "no reply
+ * parent yet").
+ */
+export const resolveReplyTargets = query({
+  args: {
+    channelId: v.id("channels"),
+    discordMessageIds: v.array(v.string()),
+  },
+  returns: v.array(
+    v.object({
+      discordMessageId: v.string(),
+      messageId: v.id("messages"),
+    }),
+  ),
+  handler: async (ctx, args) => {
+    const out: Array<{ discordMessageId: string; messageId: Id<"messages"> }> = [];
+    for (const dId of args.discordMessageIds) {
+      const row = await ctx.db
+        .query("messages")
+        .withIndex("by_channel_discord_message_id", (q) =>
+          q.eq("channelId", args.channelId).eq("discordMessageId", dId),
+        )
+        .first();
+      if (row) out.push({ discordMessageId: dId, messageId: row._id });
+    }
+    return out;
+  },
+});
+
+/**
+ * Returns the highest Discord message ID already imported into
+ * `channelId`, or null if nothing's been imported yet. The runner
+ * uses this to resume after a restart without double-uploading
+ * attachments for rows that are already in Convex.
+ *
+ * Note: Discord snowflakes are monotonic by creation time, so
+ * "highest string when compared lexicographically after left-padding
+ * to 20 chars" ≈ "most recent". We sort by `importedCreatedAt`
+ * (already indexed) instead, which is more robust and avoids
+ * scanning every row.
+ */
+export const getImportProgress = query({
+  args: { channelId: v.id("channels"), actorId: v.id("userProfiles") },
+  returns: v.object({
+    importedCount: v.number(),
+    latestImportedAt: v.union(v.number(), v.null()),
+    earliestImportedAt: v.union(v.number(), v.null()),
+    latestDiscordMessageId: v.union(v.string(), v.null()),
+  }),
+  handler: async (ctx, args) => {
+    if (!(await hasPermission(ctx, args.actorId, "manage_channels"))) {
+      throw new Error("Not authorized");
+    }
+    // Newest imported row first.
+    const latest = await ctx.db
+      .query("messages")
+      .withIndex("by_channel_imported_at", (q) =>
+        q.eq("channelId", args.channelId),
+      )
+      .order("desc")
+      .first();
+    // Oldest imported row first.
+    const earliest = await ctx.db
+      .query("messages")
+      .withIndex("by_channel_imported_at", (q) =>
+        q.eq("channelId", args.channelId),
+      )
+      .order("asc")
+      .first();
+
+    // Count via a bounded take — no dedicated counter. 10k is a
+    // generous ceiling for the progress badge; the real ground
+    // truth is the client's own run state. We return `importedCount`
+    // as an approximate "at least" measure.
+    const sample = await ctx.db
+      .query("messages")
+      .withIndex("by_channel_imported_at", (q) =>
+        q.eq("channelId", args.channelId),
+      )
+      .take(10_000);
+    const importedCount = sample.filter(
+      (m) => m.importedCreatedAt !== undefined,
+    ).length;
+
+    return {
+      importedCount,
+      latestImportedAt: latest?.importedCreatedAt ?? null,
+      earliestImportedAt: earliest?.importedCreatedAt ?? null,
+      latestDiscordMessageId: latest?.discordMessageId ?? null,
+    };
+  },
+});
+
+/**
+ * List ghost users for the admin merge UI. Each row includes a
+ * message count (bounded-take approximation) so the operator can
+ * prioritise merging noisy ghosts first.
+ */
+export const listGhosts = query({
+  args: { actorId: v.id("userProfiles") },
+  returns: v.array(
+    v.object({
+      _id: v.id("userProfiles"),
+      username: v.string(),
+      displayName: v.optional(v.string()),
+      discordId: v.optional(v.string()),
+      ghostAvatarUrl: v.optional(v.string()),
+      messageCount: v.number(),
+    }),
+  ),
+  handler: async (ctx, args) => {
+    if (!(await hasPermission(ctx, args.actorId, "manage_channels"))) {
+      throw new Error("Not authorized");
+    }
+    const users = await ctx.db.query("userProfiles").collect();
+    const ghosts = users.filter((u) => u.isGhost);
+    const out = [];
+    for (const g of ghosts) {
+      const sample = await ctx.db
+        .query("messages")
+        .withIndex("by_sender", (q) => q.eq("senderId", g._id))
+        .take(1000);
+      out.push({
+        _id: g._id,
+        username: g.username,
+        displayName: g.displayName,
+        discordId: g.discordId,
+        ghostAvatarUrl: g.ghostAvatarUrl,
+        messageCount: sample.length,
+      });
+    }
+    out.sort((a, b) => b.messageCount - a.messageCount);
+    return out;
+  },
+});
+
+/**
+ * Admin-only fetch of the current channel roster's public keys
+ * (used by the import UI to discover which local users could be
+ * mapped to Discord authors). Returns ghosts too so the UI can
+ * show "already-imported-as-ghost" inline.
+ */
+export const listMappingCandidates = query({
+  args: { actorId: v.id("userProfiles") },
+  returns: v.array(
+    v.object({
+      _id: v.id("userProfiles"),
+      username: v.string(),
+      displayName: v.optional(v.string()),
+      avatarUrl: v.union(v.string(), v.null()),
+      isGhost: v.boolean(),
+      discordId: v.optional(v.string()),
+    }),
+  ),
+  handler: async (ctx, args) => {
+    if (!(await hasPermission(ctx, args.actorId, "manage_channels"))) {
+      throw new Error("Not authorized");
+    }
+    const users = await ctx.db.query("userProfiles").collect();
+    const out = [];
+    for (const u of users) {
+      let avatarUrl: string | null = null;
+      if (u.avatarStorageId) {
+        avatarUrl = await getPublicStorageUrl(ctx, u.avatarStorageId);
+      } else if (u.isGhost && u.ghostAvatarUrl) {
+        avatarUrl = u.ghostAvatarUrl;
+      }
+      out.push({
+        _id: u._id,
+        username: u.username,
+        displayName: u.displayName,
+        avatarUrl,
+        isGhost: !!u.isGhost,
+        discordId: u.discordId,
+      });
+    }
+    return out;
+  },
+});