memind/experience-service-pg.mjs

import crypto from 'node:crypto';

import {
  experienceSearchHaystack,
  mapExperienceRow,
  normalizeExperienceRecordInput,
  serializeExperienceJson,
} from './experience-schema.mjs';

/**
 * PostgreSQL + pgvector backed experience store (etat C, polyglot mode).
 *
 * Same contract as the MySQL `createExperienceService` (record / search /
 * reflect) so callers (mindspace-agent-runner) are backend-agnostic. The MySQL
 * business DB is untouched; only this shared-experience workload lives on PG.
 *
 * Semantic search via pgvector is used when an `embed(text) => number[]` function
 * is supplied; otherwise it degrades to keyword (ILIKE) + recency, identical in
 * spirit to the MySQL version, so it is useful before an embedding provider is wired.
 *
 * The `pg` driver is imported dynamically so this module loads even when the
 * dependency is not yet installed; it is only required once the service is built.
 *
 * @param {object} options
 * @param {string} options.connectionString - e.g. process.env.EXPERIENCE_PG_URL
 * @param {(text: string) => Promise<number[]>} [options.embed] - embedding fn
 * @param {number} [options.embeddingDim=1536] - vector dimension (must match embed)
 * @param {() => number} [options.now]
 * @param {number} [options.recencyHalfLifeMs]
 */
export async function createPgExperienceService(options = {}) {
  const connectionString = options.connectionString;
  if (!connectionString) {
    throw new Error('createPgExperienceService 需要 connectionString（建议用 EXPERIENCE_PG_URL）');
  }
  const now = options.now ?? (() => Date.now());
  const embed = options.embed ?? null;
  const embeddingDim = Math.max(1, Number(options.embeddingDim ?? 1536));
  const halfLifeMs = Math.max(
    60_000,
    Number(options.recencyHalfLifeMs ?? 30 * 24 * 60 * 60 * 1000),
  );

  const { default: pg } = await import('pg');
  const pool = new pg.Pool({ connectionString, max: Number(options.poolMax ?? 10) });

  // Idempotent setup. pgvector must be available on the server (CREATE EXTENSION).
  await pool.query('CREATE EXTENSION IF NOT EXISTS vector');
  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_experience (
      id UUID PRIMARY KEY,
      scope TEXT NOT NULL DEFAULT 'global',
      kind TEXT NOT NULL DEFAULT 'lesson',
      title TEXT NOT NULL,
      body TEXT NOT NULL,
      tags JSONB NOT NULL DEFAULT '[]'::jsonb,
      source_session_id TEXT,
      source_user_id TEXT,
      use_count BIGINT NOT NULL DEFAULT 0,
      embedding vector(${embeddingDim}),
      problem TEXT,
      environment_json JSONB,
      hypothesis TEXT,
      action_json JSONB,
      result TEXT,
      confidence DOUBLE PRECISION,
      evidence_json JSONB,
      parent_experience_id UUID,
      supersedes_id UUID,
      status TEXT NOT NULL DEFAULT 'active',
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL
    )
  `);
  const pgExperienceColumns = [
    ['problem', 'TEXT'],
    ['environment_json', 'JSONB'],
    ['hypothesis', 'TEXT'],
    ['action_json', 'JSONB'],
    ['result', 'TEXT'],
    ['confidence', 'DOUBLE PRECISION'],
    ['evidence_json', 'JSONB'],
    ['parent_experience_id', 'UUID'],
    ['supersedes_id', 'UUID'],
    ["status", "TEXT NOT NULL DEFAULT 'active'"],
  ];
  for (const [column, definition] of pgExperienceColumns) {
    await pool.query(`ALTER TABLE h5_experience ADD COLUMN IF NOT EXISTS ${column} ${definition}`);
  }
  await pool.query(
    'CREATE INDEX IF NOT EXISTS idx_h5_experience_scope_updated ON h5_experience (scope, updated_at DESC)',
  );
  // Trigram-ish keyword fallback index (btree on lower(title)) is cheap; the
  // ANN vector index is created lazily once rows have embeddings.
  await pool.query(
    "CREATE INDEX IF NOT EXISTS idx_h5_experience_embedding ON h5_experience USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100)",
  ).catch(() => {
    // ivfflat needs the extension + may warn on empty table; non-fatal.
  });

  function toVectorLiteral(vec) {
    // pgvector accepts a string like '[0.1,0.2,...]'
    return `[${vec.map((n) => Number(n)).join(',')}]`;
  }

  function rowToExperience(row) {
    return mapExperienceRow({
      ...row,
      tags_json: row.tags,
    });
  }

  async function record(input) {
    const normalized = normalizeExperienceRecordInput(input);
    if (!normalized.title) throw experienceError('经验标题不能为空', 'invalid_experience_input');
    if (!normalized.body) throw experienceError('经验内容不能为空', 'invalid_experience_input');
    const id = crypto.randomUUID();
    const ts = now();
    const environmentJson = normalized.environment;
    const actionJson = normalized.action;
    const evidenceJson = normalized.evidence;
    let embedding = null;
    if (embed) {
      try {
        const vec = await embed(`${normalized.title}\n${normalized.body}`);
        if (Array.isArray(vec) && vec.length === embeddingDim) embedding = toVectorLiteral(vec);
      } catch {
        embedding = null;
      }
    }
    await pool.query(
      `INSERT INTO h5_experience
         (id, scope, kind, title, body, tags, source_session_id, source_user_id,
          use_count, embedding, problem, environment_json, hypothesis, action_json,
          result, confidence, evidence_json, parent_experience_id, supersedes_id,
          status, created_at, updated_at)
       VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8,0,$9::vector,$10,$11::jsonb,$12,$13::jsonb,
               $14,$15,$16::jsonb,$17::uuid,$18::uuid,$19,$20,$21)`,
      [
        id,
        normalized.scope,
        normalized.kind,
        normalized.title,
        normalized.body,
        JSON.stringify(normalized.tags),
        normalized.sourceSessionId,
        normalized.sourceUserId,
        embedding,
        normalized.problem,
        environmentJson == null ? null : JSON.stringify(environmentJson),
        normalized.hypothesis,
        actionJson == null ? null : JSON.stringify(actionJson),
        normalized.result,
        normalized.confidence,
        evidenceJson == null ? null : JSON.stringify(evidenceJson),
        normalized.parentExperienceId,
        normalized.supersedesId,
        normalized.status,
        ts,
        ts,
      ],
    );
    return rowToExperience({
      id,
      scope: normalized.scope,
      kind: normalized.kind,
      title: normalized.title,
      body: normalized.body,
      tags: normalized.tags,
      source_session_id: normalized.sourceSessionId,
      source_user_id: normalized.sourceUserId,
      use_count: 0,
      problem: normalized.problem,
      environment_json: environmentJson,
      hypothesis: normalized.hypothesis,
      action_json: actionJson,
      result: normalized.result,
      confidence: normalized.confidence,
      evidence_json: evidenceJson,
      parent_experience_id: normalized.parentExperienceId,
      supersedes_id: normalized.supersedesId,
      status: normalized.status,
      created_at: ts,
      updated_at: ts,
    });
  }

  async function search(query, { scope = 'global', limit = 5 } = {}) {
    const text = String(query ?? '').trim();
    if (!text) return [];
    const max = Math.max(1, limit);

    // Vector path: cosine distance ordering, recency as tiebreaker.
    if (embed) {
      try {
        const vec = await embed(text);
        if (Array.isArray(vec) && vec.length === embeddingDim) {
          const { rows } = await pool.query(
            `SELECT id, scope, kind, title, body, tags, source_session_id,
                    source_user_id, use_count, problem, environment_json, hypothesis,
                    action_json, result, confidence, evidence_json, parent_experience_id,
                    supersedes_id, status, created_at, updated_at
             FROM h5_experience
             WHERE scope = $1 AND status = 'active' AND embedding IS NOT NULL
             ORDER BY embedding <=> $2::vector, updated_at DESC
             LIMIT $3`,
            [scope, toVectorLiteral(vec), max],
          );
          if (rows.length > 0) {
            await bumpUseCount(rows.map((r) => r.id));
            return rows.map(rowToExperience);
          }
        }
      } catch {
        // fall through to keyword path
      }
    }

    // Keyword fallback: ILIKE any term, ordered by recency.
    const terms = tokenize(text);
    if (terms.length === 0) return [];
    const likeClauses = terms.map((_, i) => `(title ILIKE $${i + 2} OR body ILIKE $${i + 2})`);
    const params = [scope, ...terms.map((t) => `%${t}%`), max];
    const { rows } = await pool.query(
      `SELECT id, scope, kind, title, body, tags, source_session_id,
              source_user_id, use_count, problem, environment_json, hypothesis,
              action_json, result, confidence, evidence_json, parent_experience_id,
              supersedes_id, status, created_at, updated_at
       FROM h5_experience
       WHERE scope = $1 AND status = 'active' AND (${likeClauses.join(' OR ')})
       ORDER BY updated_at DESC
       LIMIT $${terms.length + 2}`,
      params,
    );
    if (rows.length > 0) await bumpUseCount(rows.map((r) => r.id));
    return rows.map(rowToExperience);
  }

  async function bumpUseCount(ids) {
    if (!ids.length) return;
    await pool
      .query('UPDATE h5_experience SET use_count = use_count + 1 WHERE id = ANY($1::uuid[])', [ids])
      .catch(() => {});
  }

  function tokenize(query) {
    return [
      ...new Set(
        String(query ?? '')
          .toLowerCase()
          .split(/[^\p{L}\p{N}]+/u)
          .map((t) => t.trim())
          .filter((t) => t.length >= 2),
      ),
    ];
  }

  async function reflect() {
    return { ok: false, reason: 'not_implemented' };
  }

  async function close() {
    await pool.end();
  }

  return { record, search, reflect, close };
}

function experienceError(message, code) {
  return Object.assign(new Error(message), { code });
}