memind/mindspace-conversation-package-backfill.mjs

import fs from 'node:fs/promises';
import path from 'node:path';
import { assetInternals } from './mindspace-assets.mjs';
import { pageInternals } from './mindspace-pages.mjs';
import { publicationInternals } from './mindspace-publications.mjs';
import { resolvePublishDir } from './user-publish.mjs';

const BACKFILL_LIMIT = 200;

function asNumber(value) {
  const number = Number(value);
  return Number.isFinite(number) ? number : null;
}

async function readUtf8IfExists(storageRoot, storageKey) {
  if (!storageRoot || !storageKey) return '';
  try {
    return await fs.readFile(path.resolve(storageRoot, storageKey), 'utf8');
  } catch {
    return '';
  }
}

export async function backfillConversationPackageArtifacts({
  pool,
  registry,
  storageRoot,
  h5Root,
  user,
  sessionId,
  now = Date.now(),
} = {}) {
  const userId = String(user?.id ?? '').trim();
  const normalizedSessionId = String(sessionId ?? '').trim();
  if (!pool || !registry || !userId || !normalizedSessionId) {
    return { pages: 0, publications: 0, uploads: 0 };
  }

  let pages = 0;
  const [pageRows] = await pool.query(
    `SELECT p.id, p.title, p.source_message_id, p.current_version_id,
            pv.version_no, pv.content_asset_id,
            av.mime_type, av.size_bytes, av.storage_key
     FROM h5_page_records p
     LEFT JOIN h5_page_versions pv ON pv.id = p.current_version_id AND pv.page_id = p.id
     LEFT JOIN h5_asset_versions av ON av.asset_id = pv.content_asset_id AND av.version_no = 1
     WHERE p.user_id = ? AND p.source_session_id = ? AND p.status <> 'deleted'
     ORDER BY p.updated_at DESC
     LIMIT ?`,
    [userId, normalizedSessionId, BACKFILL_LIMIT],
  );
  for (const row of pageRows) {
    const artifact = await pageInternals.registerPageArtifactForConversation({
      registry,
      userId,
      source: { sessionId: normalizedSessionId, messageId: row.source_message_id ?? null },
      page: {
        id: row.id,
        title: row.title,
        currentVersionId: row.current_version_id,
        sourceMessageId: row.source_message_id ?? null,
      },
      contentAssetId: row.content_asset_id ?? null,
      contentMimeType: row.mime_type ?? null,
      contentBytes: asNumber(row.size_bytes),
      storageKey: row.storage_key ?? null,
      versionNo: row.version_no,
      now,
    });
    if (artifact) pages += 1;
  }

  let publications = 0;
  const [publicationRows] = await pool.query(
    `SELECT pr.id, pr.page_id, pr.public_url, pr.published_at,
            p.title, p.source_message_id,
            pv.bundle_asset_id,
            JSON_UNQUOTE(JSON_EXTRACT(pv.source_snapshot_json, '$.relative_path')) AS source_relative_path,
            av.storage_key, av.size_bytes
     FROM h5_publish_records pr
     JOIN h5_page_records p ON p.id = pr.page_id AND p.user_id = pr.user_id
     JOIN h5_page_versions pv ON pv.id = pr.page_version_id AND pv.page_id = p.id
     LEFT JOIN h5_asset_versions av ON av.asset_id = pv.bundle_asset_id AND av.version_no = 1
     WHERE pr.user_id = ? AND p.source_session_id = ? AND pr.status IN ('online', 'offline')
     ORDER BY pr.published_at DESC
     LIMIT ?`,
    [userId, normalizedSessionId, BACKFILL_LIMIT],
  );
  const publishDir = h5Root ? resolvePublishDir(h5Root, { id: userId }) : null;
  for (const row of publicationRows) {
    const artifact = await publicationInternals.registerPublicationArtifactForConversation({
      registry,
      userId,
      page: {
        page_id: row.page_id,
        title: row.title,
        source_session_id: normalizedSessionId,
        source_message_id: row.source_message_id ?? null,
        source_relative_path: row.source_relative_path ?? null,
      },
      publication: {
        id: row.id,
        publicUrl: row.public_url,
        publishedAt: row.published_at,
      },
      bundleAssetId: row.bundle_asset_id ?? null,
      storageKey: row.storage_key ?? null,
      htmlBytes: asNumber(row.size_bytes),
      html: await readUtf8IfExists(storageRoot, row.storage_key),
      publishDir,
      now,
    });
    if (artifact) publications += 1;
  }

  let uploads = 0;
  const [uploadRows] = await pool.query(
    `SELECT u.id, u.filename, u.source_message_id, u.completed_asset_id,
            u.detected_mime_type, u.actual_size,
            a.mime_type, a.original_filename, a.display_name, a.size_bytes,
            v.storage_key
     FROM h5_upload_sessions u
     JOIN h5_assets a ON a.id = u.completed_asset_id AND a.user_id = u.user_id
     LEFT JOIN h5_asset_versions v ON v.id = a.current_version_id AND v.asset_id = a.id
     WHERE u.user_id = ? AND u.source_session_id = ? AND u.status = 'completed'
     ORDER BY COALESCE(u.completed_at, u.created_at) DESC
     LIMIT ?`,
    [userId, normalizedSessionId, BACKFILL_LIMIT],
  );
  for (const row of uploadRows) {
    const artifact = await assetInternals.registerUploadArtifactForConversation({
      registry,
      userId,
      upload: {
        filename: row.filename,
        source_session_id: normalizedSessionId,
        source_message_id: row.source_message_id ?? null,
        detected_mime_type: row.detected_mime_type ?? row.mime_type ?? null,
        actual_size: row.actual_size ?? row.size_bytes ?? null,
      },
      asset: {
        id: row.completed_asset_id,
        mimeType: row.mime_type ?? row.detected_mime_type ?? null,
        filename: row.original_filename ?? row.filename,
        displayName: row.display_name ?? row.filename,
        sizeBytes: asNumber(row.size_bytes ?? row.actual_size),
      },
      assetId: row.completed_asset_id,
      storageKey: row.storage_key ?? null,
      canonicalUrl: `/api/mindspace/v1/assets/${encodeURIComponent(row.completed_asset_id)}/download`,
      now,
    });
    if (artifact) uploads += 1;
  }

  return { pages, publications, uploads };
}