memind/user-model-service/ingest.mjs

import crypto from 'node:crypto';

function newId() {
  return crypto.randomUUID();
}

function nowMs() {
  return Date.now();
}

function validateEnvelope(item) {
  const required = [
    'evidence_id',
    'user_id',
    'source_type',
    'source_ref',
    'occurred_at',
    'evidence_type',
    'payload',
    'content_hash',
    'schema_version',
  ];
  for (const key of required) {
    if (item[key] === undefined || item[key] === null || item[key] === '') {
      return { ok: false, reason: `missing ${key}` };
    }
  }
  if (item.schema_version !== 1) return { ok: false, reason: 'unsupported schema_version' };
  if (item.privacy_level === 'secure_skip') return { ok: false, reason: 'secure_skip' };
  return { ok: true };
}

/**
 * @param {import('mysql2/promise').Pool} pool
 * @param {{ items: object[], dry_run?: boolean }} input
 */
export async function ingestEvidenceBatch(pool, { items, dry_run = false }) {
  const accepted = [];
  const duplicates = [];
  const rejected = [];

  for (const item of items ?? []) {
    const check = validateEnvelope(item);
    if (!check.ok) {
      rejected.push({ evidence_id: item.evidence_id ?? null, reason: check.reason });
      continue;
    }

    const [existing] = await pool.query(
      `SELECT evidence_id, content_hash FROM um_evidence WHERE user_id = ? AND content_hash = ? LIMIT 1`,
      [item.user_id, item.content_hash],
    );
    if (existing[0]) {
      duplicates.push(existing[0].evidence_id);
      continue;
    }

    const [byId] = await pool.query(`SELECT content_hash FROM um_evidence WHERE evidence_id = ? LIMIT 1`, [
      item.evidence_id,
    ]);
    if (byId[0] && byId[0].content_hash !== item.content_hash) {
      rejected.push({ evidence_id: item.evidence_id, reason: 'evidence_conflict' });
      continue;
    }

    if (dry_run) {
      accepted.push(item.evidence_id);
      continue;
    }

    await pool.query(
      `INSERT INTO um_evidence
       (evidence_id, user_id, source_type, source_ref, evidence_type, occurred_at, received_at,
        content_hash, schema_version, privacy_level, payload_json)
       VALUES (?, ?, ?, ?, ?, ?, NOW(3), ?, ?, ?, ?)`,
      [
        item.evidence_id,
        item.user_id,
        item.source_type,
        item.source_ref,
        item.evidence_type,
        item.occurred_at.replace('T', ' ').replace('Z', '').slice(0, 23),
        item.content_hash,
        item.schema_version,
        item.privacy_level ?? 'normal',
        JSON.stringify(item.payload),
      ],
    );
    accepted.push(item.evidence_id);
  }

  return { accepted, duplicates, rejected };
}

export async function updateIngestCursor(pool, userId, sourceType, cursorValue) {
  await pool.query(
    `INSERT INTO um_ingest_cursors (user_id, source_type, cursor_value, updated_at)
     VALUES (?, ?, ?, ?)
     ON DUPLICATE KEY UPDATE cursor_value = VALUES(cursor_value), updated_at = VALUES(updated_at)`,
    [userId, sourceType, cursorValue, nowMs()],
  );
}

export { validateEnvelope };