memind/user-model-service/service.mjs

import { mergeCandidatesFromSignals } from './candidates.mjs';
import { ingestEvidenceBatch, updateIngestCursor } from './ingest.mjs';
import { extractSignalsFromEnvelope, upsertSignals } from './signals.mjs';
import { materializeProfileAndSnapshot } from './snapshot.mjs';

/**
 * @param {import('mysql2/promise').Pool} pool
 * @param {{ items: object[], source_type?: string, dry_run?: boolean }} input
 */
export async function processIngestBatch(pool, input) {
  const ingestResult = await ingestEvidenceBatch(pool, input);
  if (input.dry_run) {
    return { ...ingestResult, signals_computed: 0, candidates_touched: 0, snapshot: null };
  }

  const userIds = new Set();
  for (const id of ingestResult.accepted) {
    const [rows] = await pool.query(`SELECT user_id, payload_json, evidence_type, evidence_id, occurred_at, source_type, source_ref, content_hash, schema_version, privacy_level FROM um_evidence WHERE evidence_id = ?`, [id]);
    const row = rows[0];
    if (!row) continue;
    userIds.add(row.user_id);
    const envelope = {
      evidence_id: row.evidence_id,
      user_id: row.user_id,
      source_type: row.source_type,
      source_ref: row.source_ref,
      occurred_at: row.occurred_at instanceof Date ? row.occurred_at.toISOString() : row.occurred_at,
      evidence_type: row.evidence_type,
      payload: typeof row.payload_json === 'string' ? JSON.parse(row.payload_json) : row.payload_json,
      content_hash: row.content_hash,
      schema_version: row.schema_version,
      privacy_level: row.privacy_level,
    };
    const drafts = extractSignalsFromEnvelope(envelope);
    await upsertSignals(pool, row.user_id, drafts);
  }

  let signalsComputed = 0;
  let candidatesTouched = 0;
  let snapshot = null;

  for (const userId of userIds) {
    signalsComputed += 1;
    candidatesTouched += await mergeCandidatesFromSignals(pool, userId);
    snapshot = await materializeProfileAndSnapshot(pool, userId, { reason: 'ingest_batch' });
  }

  const sourceType = input.source_type ?? input.items?.[0]?.source_type ?? 'meinput';
  const lastItem = input.items?.[input.items.length - 1];
  if (lastItem?.occurred_at && userIds.size === 1) {
    await updateIngestCursor(pool, [...userIds][0], sourceType, lastItem.occurred_at);
  }

  return {
    ...ingestResult,
    signals_computed: signalsComputed,
    candidates_touched: candidatesTouched,
    snapshot: snapshot
      ? {
          profile_version_bumped: true,
          fast_revision_bumped: true,
          ...snapshot,
        }
      : { profile_version_bumped: false, fast_revision_bumped: false },
  };
}